Upload folder using huggingface_hub

.gitattributes

@@ -33,3 +33,16 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+assistant_female_voice.wav filter=lfs diff=lfs merge=lfs -text
+models/VoxCPM-0.5B/assets/voxcpm_model.png filter=lfs diff=lfs merge=lfs -text
+models/iic/SenseVoiceSmall/fig/aed_figure.png filter=lfs diff=lfs merge=lfs -text
+models/iic/SenseVoiceSmall/fig/asr_results.png filter=lfs diff=lfs merge=lfs -text
+models/iic/SenseVoiceSmall/fig/inference.png filter=lfs diff=lfs merge=lfs -text
+models/iic/SenseVoiceSmall/fig/sensevoice.png filter=lfs diff=lfs merge=lfs -text
+models/iic/SenseVoiceSmall/fig/ser_figure.png filter=lfs diff=lfs merge=lfs -text
+models/iic/SenseVoiceSmall/fig/ser_table.png filter=lfs diff=lfs merge=lfs -text
+models/iic/speech_zipenhancer_ans_multiloss_16k_base/description/matrix.jpg filter=lfs diff=lfs merge=lfs -text
+models/iic/speech_zipenhancer_ans_multiloss_16k_base/description/matrix_voicebank.jpg filter=lfs diff=lfs merge=lfs -text
+models/iic/speech_zipenhancer_ans_multiloss_16k_base/examples/speech_with_noise1.wav filter=lfs diff=lfs merge=lfs -text
+models/v10/tokenizer.json filter=lfs diff=lfs merge=lfs -text
+spk_001.wav filter=lfs diff=lfs merge=lfs -text

Dockerfile

@@ -0,0 +1,47 @@
+FROM nvidia/cuda:12.3.2-cudnn9-devel-ubuntu22.04
+
+# Set environment variables
+ENV PYTHONUNBUFFERED=1 \
+    DEBIAN_FRONTEND=noninteractive \
+    CUDA_HOME=/usr/local/cuda \
+    PATH=/usr/local/cuda/bin:$PATH \
+    LD_LIBRARY_PATH=/usr/local/cuda/lib64:$LD_LIBRARY_PATH \
+    NVIDIA_VISIBLE_DEVICES=all \
+    NVIDIA_DRIVER_CAPABILITIES=compute,utility \
+    HF_HOME=/app/models \
+    TRITON_CACHE_DIR=/tmp/triton_cache \
+    XDG_CACHE_HOME=/tmp \
+    NUMBA_CACHE_DIR=/tmp/numba_cache
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    python3 \
+    python3-pip \
+    python3-dev \
+    build-essential \
+    git \
+    ffmpeg \
+    libsndfile1 \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Upgrade pip and install build tools
+RUN python3 -m pip install --upgrade pip setuptools wheel uv
+
+WORKDIR /app
+
+# Create Numba cache directory
+RUN mkdir -p /tmp/numba_cache /tmp/triton_cache && \
+    chown nobody:nogroup /tmp/numba_cache /tmp/triton_cache && \
+    chmod 700 /tmp/numba_cache /tmp/triton_cache
+
+COPY requirements.txt .
+
+# Install other requirements
+RUN python3 -m uv pip install --no-cache-dir -r requirements.txt --prerelease=allow
+
+COPY . .
+
+EXPOSE 8010
+
+# CMD ["python3", "server.py"]

README.md

@@ -0,0 +1,13 @@
+---
+license: mit
+tags:
+- any-to-any
+- omega
+- omegalabs
+- bittensor
+- agi
+---
+
+This is an Any-to-Any model checkpoint for the OMEGA Labs x Bittensor Any-to-Any subnet.
+
+Check out the [git repo](https://github.com/omegalabsinc/omegalabs-anytoany-bittensor) and find OMEGA on X: [@omegalabsai](https://x.com/omegalabsai).

assistant_female_voice.wav

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

attention_mask_research.md

@@ -0,0 +1,186 @@
+# Attention Masks and Pad Tokens in Transformer Generation: Research Questions
+
+## Core Problem Statement
+
+When running transformer models (specifically Llama-3.2-1B-Instruct) for text generation, we encounter warnings about missing attention masks and pad tokens, even for single input sequences. This leads to inconsistent generation outputs despite identical inputs.
+
+### Warning Messages Observed
+```
+The attention mask and the pad token id were not set. As a consequence, you may observe unexpected behavior. Please pass your input's `attention_mask` to obtain reliable results.
+Setting `pad_token_id` to `eos_token_id`:128001 for open-end generation.
+The attention mask is not set and cannot be inferred from input because pad token is same as eos token.
+```
+
+## Key Research Questions
+
+### 1. Why do single inputs require attention masks?
+**Initial Assumption**: Single sequences without padding shouldn't need attention masks.
+**Observed Reality**: Even single inputs show different generation outputs when attention masks are missing.
+
+### 2. What is the relationship between pad tokens and attention masks?
+**Question**: How do pad_token_id and attention_mask work together in the generation process?
+
+### 3. Why does pad_token_id = eos_token_id cause issues?
+**Specific Issue**: When padding token equals end-of-sequence token, what ambiguity does this create?
+
+## Code Analysis
+
+### Current Implementation (Problematic)
+```python
+def chat_current(system_prompt: str, user_prompt: str) -> str:
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_prompt},
+    ]
+
+    # Only returns input_ids tensor
+    input_ids = tok.apply_chat_template(
+        messages,
+        add_generation_prompt=True,
+        return_tensors="pt"
+    ).to(lm.device)
+
+    with torch.inference_mode():
+        output_ids = lm.generate(
+            input_ids,  # Missing: attention_mask, pad_token_id
+            max_new_tokens=2048,
+            do_sample=True,
+            temperature=0.2,
+            repetition_penalty=1.1,
+            top_k=100,
+            top_p=0.95,
+        )
+    
+    return tok.decode(output_ids[0][input_ids.shape[-1]:], skip_special_tokens=True)
+```
+
+### Fixed Implementation
+```python
+def chat_fixed(system_prompt: str, user_prompt: str) -> str:
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_prompt},
+    ]
+
+    # Returns dictionary with input_ids AND attention_mask
+    inputs = tok.apply_chat_template(
+        messages,
+        add_generation_prompt=True,
+        return_tensors="pt",
+        return_dict=True  # KEY CHANGE: Get both components
+    )
+    
+    input_ids = inputs["input_ids"].to(lm.device)
+    attention_mask = inputs["attention_mask"].to(lm.device)
+
+    with torch.inference_mode():
+        output_ids = lm.generate(
+            input_ids=input_ids,
+            attention_mask=attention_mask,  # Explicit attention guidance
+            pad_token_id=tok.eos_token_id,  # Explicit pad token
+            max_new_tokens=2048,
+            do_sample=True,
+            temperature=0.2,
+            repetition_penalty=1.1,
+            top_k=100,
+            top_p=0.95,
+        )
+    
+    return tok.decode(output_ids[0][input_ids.shape[-1]:], skip_special_tokens=True)
+```
+
+### Model and Tokenizer Setup
+```python
+model_name = "models/Llama-3.2-1B-Instruct"
+tok = AutoTokenizer.from_pretrained(model_name)
+# Critical: Set pad token if not available
+if tok.pad_token is None:
+    tok.pad_token = tok.eos_token
+
+lm = AutoModelForCausalLM.from_pretrained(
+    model_name,
+    torch_dtype=torch.bfloat16,
+    device_map="cuda",
+).eval()
+```
+
+## Observed Behavioral Differences
+
+### Input Structure Analysis
+```python
+# Single input contains multiple components:
+messages = [
+    {"role": "system", "content": "You are a helpful assistant..."},
+    {"role": "user", "content": "What is the capital of France?"},
+]
+
+# After apply_chat_template, becomes token sequence:
+# [system_tokens, user_tokens, assistant_start_token]
+```
+
+## Technical Hypotheses for Investigation
+
+### Hypothesis 1: Internal Masking Ambiguity
+When attention_mask is missing, the model cannot distinguish between:
+- Real input tokens that should influence generation
+- Structural tokens (system prompts, role markers)
+- Token boundaries between different message roles
+
+### Hypothesis 2: EOS Token Dual Purpose Confusion  
+When `pad_token_id == eos_token_id`, the model faces ambiguity:
+```python
+# Same token (128001) serves dual purposes:
+# 1. End of sequence marker
+# 2. Padding token for batch processing
+# Model cannot infer which purpose applies in context
+```
+
+### Hypothesis 3: Autoregressive Generation Context Boundary Issues
+During generation, model needs to know:
+- Which input tokens provide valid context for next token prediction
+- Where the "prompt" ends and "generation" begins
+- How to weight attention across different input components
+
+## Research Objectives
+
+### Primary Questions
+1. **Mechanism Analysis**: How exactly does missing attention_mask affect the internal attention computation?
+2. **Consistency Impact**: Why do identical inputs produce different outputs without proper masking?
+3. **Single vs Batch Behavior**: What differences exist between single sequence and batched sequence processing?
+
+### Secondary Questions  
+1. **Model-Specific Behavior**: Do different transformer architectures handle missing attention masks differently?
+2. **Generation Parameter Interaction**: How do attention mask issues interact with sampling parameters (temperature, top_p, etc.)?
+3. **Performance Impact**: What computational overhead does proper attention masking add?
+
+## Key Technical Areas for Deep Research
+
+### Attention Mechanism Internals
+- How attention weights are computed with/without explicit masks
+- Impact on multi-head attention distributions
+- Interaction with causal masking in autoregressive models
+
+### Tokenizer Behavior
+- How `apply_chat_template` constructs input sequences
+- Default attention mask generation behavior
+- Role of special tokens in attention computation
+
+### Generation Process
+- How `model.generate()` handles missing parameters
+- Internal assumptions and fallback behaviors
+- Impact on sampling and beam search algorithms
+
+## Expected Research Outcomes
+
+Understanding of:
+1. Exact mechanism causing output inconsistency
+2. Best practices for single sequence generation  
+3. Relationship between attention masking and generation quality
+4. Guidelines for production transformer deployment
+
+## References for Deep Research
+
+- Hugging Face Transformers documentation on attention masks
+- Technical blogs on transformer attention mechanisms (2024)
+- Community discussions on pad token vs attention mask differences
+- Official model documentation for Llama architecture attention handling

compare_generation.py

@@ -0,0 +1,129 @@
+
+import torch
+from transformers import AutoModelForCausalLM, AutoTokenizer
+
+# Load model and tokenizer (same as server.py)
+model_name = "models/Llama-3.2-1B-Instruct"
+tok = AutoTokenizer.from_pretrained(model_name)
+lm = AutoModelForCausalLM.from_pretrained(
+    model_name,
+    torch_dtype=torch.bfloat16,
+    device_map="cuda",
+).eval()
+
+def chat_current(system_prompt: str, user_prompt: str) -> str:
+    """
+    Current implementation (same as server.py) - will show warnings
+    """
+    print("🔴 Running CURRENT implementation (with warnings)...")
+    
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_prompt},
+    ]
+
+
+    input_ids = tok.apply_chat_template(
+        messages,
+        add_generation_prompt=True,
+        return_tensors="pt"
+    ).to(lm.device)
+
+    with torch.inference_mode():
+        output_ids = lm.generate(
+            input_ids,  # No attention_mask, no pad_token_id
+            max_new_tokens=2048,
+            do_sample=True,
+            temperature=0.2,
+            repetition_penalty=1.1,
+            top_k=100,
+            top_p=0.95,
+        )
+
+    answer = tok.decode(
+        output_ids[0][input_ids.shape[-1]:],
+        skip_special_tokens=True,
+        clean_up_tokenization_spaces=True,
+    )
+    return answer.strip()
+
+
+def chat_fixed(system_prompt: str, user_prompt: str) -> str:
+    """
+    Fixed implementation - proper attention mask and pad token
+    """
+    print("🟢 Running FIXED implementation (no warnings)...")
+    
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_prompt},
+    ]
+
+    # Get both input_ids and attention_mask
+    inputs = tok.apply_chat_template(
+        messages,
+        add_generation_prompt=True,
+        return_tensors="pt",
+        return_dict=True  # Returns dict with input_ids and attention_mask
+    )
+    
+    # Move to device
+    input_ids = inputs["input_ids"].to(lm.device)
+    attention_mask = inputs["attention_mask"].to(lm.device)
+
+    with torch.inference_mode():
+        output_ids = lm.generate(
+            input_ids=input_ids,
+            attention_mask=attention_mask,  # Proper attention mask
+            pad_token_id=tok.eos_token_id,  # Explicit pad token
+            max_new_tokens=2048,
+            do_sample=True,
+            temperature=0.2,
+            repetition_penalty=1.1,
+            top_k=100,
+            top_p=0.95,
+        )
+
+    answer = tok.decode(
+        output_ids[0][input_ids.shape[-1]:],
+        skip_special_tokens=True,
+        clean_up_tokenization_spaces=True,
+    )
+    return answer.strip()
+
+
+def compare_generations():
+    """Compare both implementations"""
+    system_prompt = "You are a helpful assistant who tries to help answer the user's question."
+    user_prompt = "Create a report on anxiety in work. How do I manage time and stress effectively?"
+    
+    print("=" * 60)
+    print("COMPARING GENERATION METHODS")
+    print("=" * 60)
+    print(f"System: {system_prompt}")
+    print(f"User: {user_prompt}")
+    print("=" * 60)
+    
+    # Test current implementation
+    print("\n" + "=" * 60)
+    current_output = chat_current(system_prompt, user_prompt)
+    print(f"CURRENT OUTPUT:\n{current_output}")
+    
+    print("\n" + "=" * 60)
+    # Test fixed implementation  
+    fixed_output = chat_fixed(system_prompt, user_prompt)
+    print(f"FIXED OUTPUT:\n{fixed_output}")
+    
+    print("\n" + "=" * 60)
+    print("COMPARISON:")
+    print(f"Outputs are identical: {current_output == fixed_output}")
+    print(f"Current length: {len(current_output)} chars")
+    print(f"Fixed length: {len(fixed_output)} chars")
+
+
+if __name__ == "__main__":
+    # Set pad token for the fixed version
+    if tok.pad_token is None:
+        tok.pad_token = tok.eos_token
+    
+    compare_generations()

hotkey.txt

@@ -0,0 +1 @@
+5DJMUw8aibqe5BkyF33m87hBrWKuxkiuGPFxdVGzva6DSFLK

model/data.py

@@ -0,0 +1,76 @@
+from typing import Any, ClassVar, Dict, Optional, Type
+# from transformers import PreTrainedModel, PreTrainedTokenizerBase
+from pydantic import BaseModel, Field, PositiveInt, ConfigDict
+
+# The maximum bytes for metadata on the chain.
+MAX_METADATA_BYTES = 128
+# The length, in bytes, of a git commit hash.
+GIT_COMMIT_LENGTH = 40
+# The length, in bytes, of a base64 encoded sha256 hash.
+SHA256_BASE_64_LENGTH = 44
+# The max length, in characters, of the competition id
+MAX_COMPETITION_ID_LENGTH = 2
+
+
+class ModelId(BaseModel):
+    """Uniquely identifies a trained model"""
+
+    MAX_REPO_ID_LENGTH: ClassVar[int] = (
+        MAX_METADATA_BYTES
+        - GIT_COMMIT_LENGTH
+        - SHA256_BASE_64_LENGTH
+        - MAX_COMPETITION_ID_LENGTH
+        - 4  # separators
+    )
+
+    namespace: str = Field(
+        description="Namespace where the model can be found. ex. Hugging Face username/org."
+    )
+    name: str = Field(description="Name of the model.")
+
+    epoch: str = Field(description="The epoch number to submit as your checkpoint to evaluate e.g. 10")
+
+    # When handling a model locally the commit and hash are not necessary.
+    # Commit must be filled when trying to download from a remote store.
+    commit: Optional[str] = Field(
+        description="Commit of the model. May be empty if not yet committed."
+    )
+    # Hash is filled automatically when uploading to or downloading from a remote store.
+    hash: Optional[str] = Field(description="Hash of the trained model.")
+    # Identifier for competition
+    competition_id: Optional[str] = Field(description="The competition id")
+
+    def to_compressed_str(self) -> str:
+        """Returns a compressed string representation."""
+        return f"{self.namespace}:{self.name}:{self.epoch}:{self.commit}:{self.hash}:{self.competition_id}"
+
+    @classmethod
+    def from_compressed_str(cls, cs: str) -> Type["ModelId"]:
+        """Returns an instance of this class from a compressed string representation"""
+        tokens = cs.split(":")
+        return cls(
+            namespace=tokens[0],
+            name=tokens[1],
+            epoch=tokens[2] if tokens[2] != "None" else None,
+            commit=tokens[3] if tokens[3] != "None" else None,
+            hash=tokens[4] if tokens[4] != "None" else None,
+            competition_id=(
+                tokens[5] if len(tokens) >= 6 and tokens[5] != "None" else None
+            ),
+        )
+
+
+class Model(BaseModel):
+    """Represents a pre trained foundation model."""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    id: ModelId = Field(description="Identifier for this model.")
+    local_repo_dir: str = Field(description="Local repository with the required files.")
+
+
+class ModelMetadata(BaseModel):
+    id: ModelId = Field(description="Identifier for this trained model.")
+    block: PositiveInt = Field(
+        description="Block on which this model was claimed on the chain."
+    )

model/model_tracker.py

@@ -0,0 +1,224 @@
+import copy
+import datetime
+import threading
+from typing import Dict, List, Optional, Set
+import pickle
+import bittensor as bt
+import hashlib
+
+from model.data import ModelMetadata
+
+
+class NoopLock:
+    def __enter__(self):
+        pass
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        pass
+
+
+class ModelTracker:
+    """Tracks the current model for each miner.
+
+    Thread safe.
+    """
+
+    def __init__(
+        self,
+        thread_safe: bool = True,
+    ):
+        # Create a dict from miner hotkey to model metadata.
+        self.miner_hotkey_to_model_metadata_dict: dict[str, ModelMetadata] = dict()
+        # Create a dict from miner hotkey to last time it was evaluated/loaded/updated
+        self.miner_hotkey_to_last_touched_dict: dict[str, datetime.datetime] = dict()
+        # Create a dict from miner hotkey to model hash.
+        self.miner_hotkey_to_model_hash_dict: dict[str, str] = dict()
+
+        # List of overwritten models that may be safe to delete if not curently in use.
+        self.old_model_metadata: list[tuple[str, ModelMetadata]] = []
+        # List of model metadata that are currently in use.
+        self.model_metadata_in_use: set[tuple[str, str]] = set()
+
+        # Make this class thread safe because it will be accessed by multiple threads.
+        # One for the downloading new models loop and one for the validating models loop.
+        self.lock = threading.RLock() if thread_safe else NoopLock()
+
+    def save_state(self, filepath):
+        """Save the current state to the provided filepath."""
+
+        # Open a writable binary file for pickle.
+        with self.lock:
+            with open(filepath, "wb") as f:
+                pickle.dump(self.miner_hotkey_to_model_metadata_dict, f)
+
+    def load_state(self, filepath):
+        """Load the state from the provided filepath."""
+
+        # Open a readable binary file for pickle.
+        with open(filepath, "rb") as f:
+            self.miner_hotkey_to_model_metadata_dict = pickle.load(f)
+
+    def get_miner_hotkey_to_model_metadata_dict(self) -> Dict[str, ModelMetadata]:
+        """Returns the mapping from miner hotkey to model metadata."""
+
+        # Return a copy to ensure outside code can't modify the scores.
+        with self.lock:
+            return copy.deepcopy(self.miner_hotkey_to_model_metadata_dict)
+
+    def get_model_metadata_for_miner_hotkey(
+        self, hotkey: str
+    ) -> Optional[ModelMetadata]:
+        """Returns the model metadata for a given hotkey if any."""
+
+        with self.lock:
+            if hotkey in self.miner_hotkey_to_model_metadata_dict:
+                return self.miner_hotkey_to_model_metadata_dict[hotkey]
+            return None
+
+    def take_model_metadata_for_miner_hotkey(self, hotkey: str) -> Optional[ModelMetadata]:
+        """Returns the model metadata for a given hotkey if any. Also, marks it as in use to prevent race conditions."""
+
+        with self.lock:
+            if hotkey in self.miner_hotkey_to_model_metadata_dict:
+                metadata = self.miner_hotkey_to_model_metadata_dict[hotkey]
+                self.model_metadata_in_use.add((hotkey, metadata.id.hash))
+                return metadata
+            return None
+
+    def release_all(self):
+        with self.lock:
+            self.model_metadata_in_use.clear()
+
+    def release_model_metadata_for_miner_hotkey(self, hotkey: str, metadata: ModelMetadata):
+        with self.lock:
+            pair = (hotkey, metadata.id.hash)
+            if pair not in self.model_metadata_in_use:
+                bt.logging.error("Model metadata is not in use!")
+
+            if (hotkey, metadata) in self.old_model_metadata:
+                bt.logging.trace(f"Releasing old model metadata for hotkey: {hotkey}")
+
+            self.model_metadata_in_use.remove(pair)
+
+    def get_miner_hotkey_to_last_touched_dict(self) -> Dict[str, datetime.datetime]:
+        """Returns the mapping from miner hotkey to last time it was touched."""
+
+        # Return a copy to ensure outside code can't modify the scores.
+        with self.lock:
+            return copy.deepcopy(self.miner_hotkey_to_last_touched_dict)
+
+    def on_hotkeys_updated(self, incoming_hotkeys: Set[str]):
+        """Notifies the tracker which hotkeys are currently being tracked on the metagraph."""
+
+        with self.lock:
+            existing_hotkeys = set(self.miner_hotkey_to_model_metadata_dict.keys())
+            for hotkey in existing_hotkeys - incoming_hotkeys:
+                del self.miner_hotkey_to_model_metadata_dict[hotkey]
+                bt.logging.trace(f"Removed outdated hotkey metadata: {hotkey} from ModelTracker")
+
+            existing_hotkeys = set(self.miner_hotkey_to_last_touched_dict.keys())
+            for hotkey in existing_hotkeys - incoming_hotkeys:
+                del self.miner_hotkey_to_last_touched_dict[hotkey]
+                bt.logging.trace(f"Removed outdated hotkey timestamp: {hotkey} from ModelTracker")
+
+    def get_and_clear_old_models(self) -> list[tuple[str, ModelMetadata]]:
+        with self.lock:
+            to_delete = []
+            still_in_use = []
+            for hotkey, model in self.old_model_metadata:
+                if (hotkey, model.id.hash) in self.model_metadata_in_use:
+                    still_in_use.append((hotkey, model))
+                else:
+                    to_delete.append((hotkey, model))
+            self.old_model_metadata = still_in_use
+
+        return to_delete
+
+    def on_miner_model_updated(
+        self,
+        hotkey: str,
+        model_metadata: ModelMetadata,
+    ) -> None:
+        """Notifies the tracker that a miner has had their associated model updated.
+
+        Args:
+            hotkey (str): The miner's hotkey.
+            model_metadata (ModelMetadata): The latest model metadata of the miner.
+        """
+        with self.lock:
+            if hotkey in self.miner_hotkey_to_model_metadata_dict:
+                old_metadata = self.miner_hotkey_to_model_metadata_dict[hotkey]
+                self.old_model_metadata.append((hotkey, old_metadata))
+
+            self.miner_hotkey_to_model_metadata_dict[hotkey] = model_metadata
+            self.miner_hotkey_to_last_touched_dict[hotkey] = datetime.datetime.now()
+
+            bt.logging.trace(f"Updated Miner {hotkey}. ModelMetadata={model_metadata}.")
+
+    def touch_miner_model(self, hotkey: str) -> None:
+        """Notifies the tracker that a miner has been touched."""
+
+        now = datetime.datetime.now()
+        with self.lock:
+            self.miner_hotkey_to_last_touched_dict[hotkey] = now
+
+            bt.logging.trace(f"Touched Miner {hotkey}. datetime={now}.")
+
+    def touch_all_miner_models(self) -> None:
+        """Touch all miner models."""
+
+        now = datetime.datetime.now()
+        with self.lock:
+            for hotkey in list(self.miner_hotkey_to_model_metadata_dict.keys()):
+                self.miner_hotkey_to_last_touched_dict[hotkey] = now
+
+            bt.logging.trace(f"Touched All Miners. datetime={now}.")
+
+    def update_model_hash(self, hotkey: str, new_model_hash: str) -> bool:
+        """
+        Update the model_hash for a given hotkey.
+        
+        Args:
+        hotkey (str): The miner's hotkey.
+        new_model_hash (str): The new model hash to be set.
+        
+        Returns:
+        bool: True if the update was successful, False if the hotkey was not found.
+        """
+        with self.lock:
+            self.miner_hotkey_to_model_hash_dict[hotkey] = new_model_hash
+            return True
+
+    def calculate_file_hash(self, file_path: str) -> str:
+        """Calculate SHA1 hash of a file."""
+        sha1 = hashlib.sha1()
+        with open(file_path, 'rb') as f:
+            while True:
+                data = f.read(65536)  # Read in 64kb chunks
+                if not data:
+                    break
+                sha1.update(data)
+        return sha1.hexdigest()
+
+    def is_model_unique(self, hotkey_to_check: str, block_to_check: int, model_checkpoint_path: str) -> bool:
+        """Check if a model with a given model_hash is already in use."""
+        # generate hash from model_checkpoint_path
+        hash_to_check = self.calculate_file_hash(model_checkpoint_path)
+
+        with self.lock:
+            for hotkey, metadata in self.miner_hotkey_to_model_metadata_dict.items():
+                if hotkey == hotkey_to_check or hotkey not in self.miner_hotkey_to_model_hash_dict:
+                    continue
+                
+                if self.miner_hotkey_to_model_hash_dict[hotkey] == hash_to_check and metadata.block < block_to_check:
+                    bt.logging.warning(
+                        f"*** Model with hash {hash_to_check} on block {block_to_check} is not unique. Already in use by {hotkey} on block {metadata.block} for model {metadata.id.namespace}/{metadata.id.name}. ***"
+                    )
+                    # Update the model hash for the hotkey
+                    self.update_model_hash(hotkey_to_check, hash_to_check)
+                    return False, hash_to_check
+                
+            # Update the model hash for the hotkey
+            self.update_model_hash(hotkey_to_check, hash_to_check)
+            return True, hash_to_check
+        

model/model_updater.py

@@ -0,0 +1,93 @@
+import bittensor as bt
+from typing import Optional
+from constants import CompetitionParameters, COMPETITION_SCHEDULE
+import constants
+from model.data import ModelMetadata, Model
+from model.model_tracker import ModelTracker
+from model.storage.local_model_store import LocalModelStore
+from model.storage.model_metadata_store import ModelMetadataStore
+from model.storage.remote_model_store import RemoteModelStore
+
+
+class ModelUpdater:
+    """Checks if the currently tracked model for a hotkey matches what the miner committed to the chain."""
+
+    def __init__(
+        self,
+        metadata_store: ModelMetadataStore,
+        remote_store: RemoteModelStore,
+        local_store: LocalModelStore,
+        model_tracker: ModelTracker,
+    ):
+        self.metadata_store = metadata_store
+        self.remote_store = remote_store
+        self.local_store = local_store
+        self.model_tracker = model_tracker
+        self.min_block: Optional[int] = None
+
+    def set_min_block(self, val: Optional[int]):
+        self.min_block = val
+
+    @classmethod
+    def get_competition_parameters(cls, id: str) -> Optional[CompetitionParameters]:
+        for x in COMPETITION_SCHEDULE:
+            if x.competition_id == id:
+                return x
+        return None
+
+    async def _get_metadata(self, hotkey: str) -> Optional[ModelMetadata]:
+        """Get metadata about a model by hotkey"""
+        return await self.metadata_store.retrieve_model_metadata(hotkey)
+
+    async def sync_model(self, hotkey: str) -> bool:
+        """Updates local model for a hotkey if out of sync and returns if it was updated."""
+        # Get the metadata for the miner.
+        metadata = await self._get_metadata(hotkey)
+
+        if not metadata:
+            bt.logging.trace(
+                f"No valid metadata found on the chain for hotkey {hotkey}"
+            )
+            return False
+
+        if self.min_block and metadata.block < self.min_block:
+            bt.logging.trace(
+                f"Skipping model for {hotkey} since it was submitted at block {metadata.block} which is less than the minimum block {self.min_block}"
+            )
+            return False
+
+        # Backwards compatability for models submitted before competition id added
+        if metadata.id.competition_id is None:
+            metadata.id.competition_id = constants.ORIGINAL_COMPETITION_ID
+
+        parameters = ModelUpdater.get_competition_parameters(metadata.id.competition_id)
+        if not parameters:
+            bt.logging.trace(
+                f"No competition parameters found for {metadata.id.competition_id}"
+            )
+            return False
+
+        # Check what model id the model tracker currently has for this hotkey.
+        tracker_model_metadata = self.model_tracker.get_model_metadata_for_miner_hotkey(
+            hotkey
+        )
+        if metadata == tracker_model_metadata:
+            return False
+        bt.logging.debug(f"Syncing model for hotkey {hotkey}")
+        # Get the local path based on the local store to download to (top level hotkey path)
+        path = self.local_store.get_path(hotkey)
+        
+        # bt.logging.warning(f"Downloading model to {path}")
+        # # Otherwise we need to download the new model based on the metadata.
+        # model = await self.remote_store.download_model(metadata.id, path, parameters)
+        # bt.logging.warning(f"Downloaded model to {path}")
+        # # Check that the hash of the downloaded content matches.
+        # if model.id.hash != metadata.id.hash:
+        #     raise ValueError(
+        #         f"Sync for hotkey {hotkey} failed. Hash of content downloaded from hugging face does not match chain metadata. {metadata}"
+        #     )
+
+        # Update the tracker
+        self.model_tracker.on_miner_model_updated(hotkey, metadata)
+        bt.logging.info(f"Model for hotkey {hotkey} updated to {metadata}")
+        return True

model/storage/chain/chain_model_metadata_store.py

@@ -0,0 +1,177 @@
+import asyncio
+import functools
+import bittensor as bt
+import os
+from model.data import ModelId, ModelMetadata
+import constants
+from model.storage.model_metadata_store import ModelMetadataStore
+from typing import Optional
+
+from utilities import utils
+
+
+class ChainModelMetadataStore(ModelMetadataStore):
+    """Chain based implementation for storing and retrieving metadata about a model."""
+
+    def __init__(
+        self,
+        subtensor: bt.subtensor,
+        subnet_uid: int,
+        wallet: Optional[bt.wallet] = None,
+    ):
+        self.subtensor = subtensor
+        self.wallet = (
+            wallet  # Wallet is only needed to write to the chain, not to read.
+        )
+        self.subnet_uid = subnet_uid
+
+        # this is a hacky way to prime the get_metadata function
+        SN21_OWNER_KEY = "5GsHpHeCGhxstoEEZTR64VUashnDP4n7ir7LbNdRfXpkMU7R"
+        metadata = bt.extrinsics_subpackage.serving.get_metadata(self.subtensor, self.subnet_uid, SN21_OWNER_KEY)
+        bt.logging.debug(f"primed get_metadata call successfully: metadata={metadata} (ok to be None)")
+
+    async def store_model_metadata(self, hotkey: str, model_id: ModelId):
+        """Stores model metadata on this subnet for a specific wallet."""
+        if self.wallet is None:
+            raise ValueError("No wallet available to write to the chain.")
+
+        # Wrap calls to the subtensor in a subprocess with a timeout to handle potential hangs.
+        # partial = functools.partial(
+        #     self.subtensor.commit,
+        #     self.wallet,
+        #     self.subnet_uid,
+        #     model_id.to_compressed_str(),
+        # )
+        # utils.run_in_subprocess(partial, 60)
+        commit = self.subtensor.commit(self.wallet, self.subnet_uid, model_id.to_compressed_str())
+        print(f"success commit: {commit}")
+        if not commit:
+            raise ValueError("Failed to commit model metadata to the chain.")
+        return commit
+
+    async def retrieve_model_metadata(self, hotkey: str) -> Optional[ModelMetadata]:
+        """Retrieves model metadata on this subnet for specific hotkey"""
+
+        # Wrap calls to the subtensor in a subprocess with a timeout to handle potential hangs.
+        partial = functools.partial(
+            bt.extrinsics_subpackage.serving.get_metadata, self.subtensor, self.subnet_uid, hotkey
+        )
+
+        metadata = utils.run_in_subprocess(partial, 60)
+
+        if not metadata:
+            return None
+
+        commitment = metadata["info"]["fields"][0][0]
+
+        hex_data_tuple = commitment[list(commitment.keys())[0]][0]
+
+        chain_str = ''.join(chr(num) for num in hex_data_tuple)
+
+        model_id = None
+
+        try:
+            model_id = ModelId.from_compressed_str(chain_str)
+        except:
+            # If the metadata format is not correct on the chain then we return None.
+            bt.logging.trace(
+                f"Failed to parse the metadata on the chain for hotkey {hotkey}."
+            )
+            return None
+
+        model_metadata = ModelMetadata(id=model_id, block=metadata["block"])
+        return model_metadata
+
+
+# Can only commit data every ~20 minutes.
+async def test_store_model_metadata():
+    """Verifies that the ChainModelMetadataStore can store data on the chain."""
+    model_id = ModelId(
+        namespace="TestPath", name="TestModel", hash="TestHash1", commit="1.0"
+    )
+
+    # Use a different subnet that does not leverage chain storage to avoid conflicts.
+    # TODO switch to a mocked version when it supports commits.
+    subtensor = bt.subtensor()
+
+    # Uses .env configured wallet/hotkey/uid for the test.
+    coldkey = os.getenv("TEST_COLDKEY")
+    hotkey = os.getenv("TEST_HOTKEY")
+    net_uid = int(os.getenv("TEST_SUBNET_UID"))
+
+    wallet = bt.wallet(name=coldkey, hotkey=hotkey)
+
+    metadata_store = ChainModelMetadataStore(
+        subtensor=subtensor, wallet=wallet, subnet_uid=net_uid
+    )
+
+    # Store the metadata on chain.
+    await metadata_store.store_model_metadata(hotkey=hotkey, model_id=model_id)
+
+    print(f"Finished storing {model_id} on the chain.")
+
+
+async def test_retrieve_model_metadata():
+    """Verifies that the ChainModelMetadataStore can retrieve data from the chain."""
+    expected_model_id = ModelId(
+        namespace="TestPath", name="TestModel", hash="TestHash1", commit="1.0"
+    )
+
+    # Use a different subnet that does not leverage chain storage to avoid conflicts.
+    # TODO switch to a mocked version when it supports commits.
+    subtensor = bt.subtensor()
+
+    # Uses .env configured hotkey/uid for the test.
+    net_uid = int(os.getenv("TEST_SUBNET_UID"))
+    hotkey_address = os.getenv("TEST_HOTKEY_ADDRESS")
+
+    # Do not require a wallet for retrieving data.
+    metadata_store = ChainModelMetadataStore(
+        subtensor=subtensor, wallet=None, subnet_uid=net_uid
+    )
+
+    # Retrieve the metadata from the chain.
+    model_metadata = await metadata_store.retrieve_model_metadata(hotkey_address)
+
+    print(f"Expecting matching model id: {expected_model_id == model_metadata.id}")
+
+
+# Can only commit data every ~20 minutes.
+async def test_roundtrip_model_metadata():
+    """Verifies that the ChainModelMetadataStore can roundtrip data on the chain."""
+    model_id = ModelId(
+        namespace="TestPath", name="TestModel", hash="TestHash1", commit="1.0"
+    )
+
+    # Use a different subnet that does not leverage chain storage to avoid conflicts.
+    # TODO switch to a mocked version when it supports commits.
+    subtensor = bt.subtensor()
+
+    # Uses .env configured wallet/hotkey/uid for the test.
+    coldkey = os.getenv("TEST_COLDKEY")
+    hotkey = os.getenv("TEST_HOTKEY")
+    net_uid = int(os.getenv("TEST_SUBNET_UID"))
+
+    wallet = bt.wallet(name=coldkey, hotkey=hotkey)
+
+    metadata_store = ChainModelMetadataStore(
+        subtensor=subtensor, wallet=wallet, subnet_uid=net_uid
+    )
+
+    # Store the metadata on chain.
+    await metadata_store.store_model_metadata(hotkey=hotkey, model_id=model_id)
+
+    # May need to use the underlying publish_metadata function with wait_for_inclusion: True to pass here.
+    # Otherwise it defaults to False and we only wait for finalization not necessarily inclusion.
+
+    # Retrieve the metadata from the chain.
+    model_metadata = await metadata_store.retrieve_model_metadata(hotkey)
+
+    print(f"Expecting matching metadata: {model_id == model_metadata.id}")
+
+
+if __name__ == "__main__":
+    # Can only commit data every ~20 minutes.
+    # asyncio.run(test_roundtrip_model_metadata())
+    # asyncio.run(test_store_model_metadata())
+    asyncio.run(test_retrieve_model_metadata())

model/storage/disk/disk_model_store.py

@@ -0,0 +1,124 @@
+import bittensor as bt
+import datetime
+import os
+from typing import Dict
+from constants import CompetitionParameters
+from model.data import Model, ModelId
+from model.storage.disk import utils
+from model.storage.local_model_store import LocalModelStore
+from transformers import AutoModelForCausalLM, AutoTokenizer
+from pathlib import Path
+
+
+class DiskModelStore(LocalModelStore):
+    """Local storage based implementation for storing and retrieving a model on disk."""
+
+    def __init__(self, base_dir: str):
+        self.base_dir = base_dir
+        os.makedirs(utils.get_local_miners_dir(base_dir), exist_ok=True)
+
+    def get_path(self, hotkey: str) -> str:
+        """Returns the path to where this store would locate this hotkey."""
+        return utils.get_local_miner_dir(self.base_dir, hotkey)
+
+    def store_model(self, hotkey: str, model: Model, hf_model: AutoModelForCausalLM, hf_tokenizer: AutoTokenizer ) -> ModelId:
+        """Stores a trained model locally."""
+        # get the path to where the model should be stored
+        model_dir = os.path.join(self.get_path(hotkey), model.id.name)
+        hf_model.save_pretrained(model_dir)
+        hf_tokenizer.save_pretrained(model_dir)
+        model.local_repo_dir = model_dir
+        
+        return model.id
+
+
+    def retrieve_model(
+        self, hotkey: str, model_id: ModelId, model_parameters: CompetitionParameters
+    ) -> Model:
+        """Retrieves a trained model locally."""
+
+        # get the path to where the model should be stored
+        model_dir = os.path.join(self.get_path(hotkey), model_id.name)
+        return Model(id=model_id, local_repo_dir=model_dir)
+
+    def delete_unreferenced_models(
+        self,
+        valid_models_by_hotkey: Dict[str, ModelId],
+        model_touched_by_hotkey: Dict[str, datetime.datetime],
+        grace_period_seconds: int,
+    ):
+        """Check across all of local storage and delete unreferenced models out of grace period."""
+        # TODO: THIS METHOD IS NOT UP TO DATE YET
+        raise NotImplementedError("This method is not implemented yet.")
+        # Expected directory structure is as follows.
+        # self.base_dir/models/hotkey/models--namespace--name/snapshots/commit/config.json + other files.
+
+        # Create a set of valid model paths up to where we expect to see the actual files.
+        valid_model_paths = set()
+        for hotkey, model_id in valid_models_by_hotkey.items():
+            valid_model_paths.add(
+                utils.get_local_model_snapshot_dir(self.base_dir, hotkey, model_id)
+            )
+
+        # For each hotkey path on disk using listdir to go one level deep.
+        miners_dir = Path(utils.get_local_miners_dir(self.base_dir))
+        hotkey_subfolder_names = [d.name for d in miners_dir.iterdir() if d.is_dir()]
+
+        for hotkey in hotkey_subfolder_names:
+            # Reconstruct the path from the hotkey
+            hotkey_path = utils.get_local_miner_dir(self.base_dir, hotkey)
+
+            # If it is not in valid_hotkeys and out of grace period remove it.
+            if hotkey not in valid_models_by_hotkey:
+                deleted_hotkey = utils.remove_dir_out_of_grace(
+                    hotkey_path, grace_period_seconds
+                )
+                if deleted_hotkey:
+                    bt.logging.trace(
+                        f"Removed directory for unreferenced hotkey: {hotkey}."
+                    )
+            else:
+                # Check all the models--namespace--name subfolder paths.
+                hotkey_dir = Path(hotkey_path)
+                model_subfolder_paths = [
+                    str(d) for d in hotkey_dir.iterdir() if d.is_dir()
+                ]
+
+                # Check all the snapshots subfolder paths
+                for model_path in model_subfolder_paths:
+                    model_dir = Path(model_path)
+                    snapshot_subfolder_paths = [
+                        str(d) for d in model_dir.iterdir() if d.is_dir()
+                    ]
+
+                    # Check all the commit paths.
+                    for snapshot_path in snapshot_subfolder_paths:
+                        snapshot_dir = Path(snapshot_path)
+                        commit_subfolder_paths = [
+                            str(d) for d in snapshot_dir.iterdir() if d.is_dir()
+                        ]
+
+                        # Reached the end. Check all the actual commit subfolders for the files.
+                        for commit_path in commit_subfolder_paths:
+                            if commit_path not in valid_model_paths:
+                                deleted_model = utils.remove_dir_out_of_grace(
+                                    commit_path, grace_period_seconds
+                                )
+                                if deleted_model:
+                                    bt.logging.trace(
+                                        f"Removing directory for unreferenced model at: {commit_path}."
+                                    )
+                            else:
+                                last_touched = model_touched_by_hotkey.get(hotkey)
+                                if last_touched is not None:
+                                    deleted_model = (
+                                        utils.remove_dir_out_of_grace_by_datetime(
+                                            commit_path,
+                                            grace_period_seconds,
+                                            last_touched,
+                                        )
+                                    )
+                                    if deleted_model:
+                                        bt.logging.trace(
+                                            f"Removing directory for stale model at: {commit_path}."
+                                        )

model/storage/disk/utils.py

@@ -0,0 +1,121 @@
+import base64
+import datetime
+import hashlib
+import os
+import shutil
+import sys
+from model.data import ModelId
+
+
+def get_local_miners_dir(base_dir: str) -> str:
+    return os.path.join(base_dir, "models")
+
+
+def get_local_miner_dir(base_dir: str, hotkey: str) -> str:
+    return os.path.join(get_local_miners_dir(base_dir), hotkey)
+
+
+# Hugging face stores models under models--namespace--name/snapshots/commit when downloading.
+def get_local_model_dir(base_dir: str, hotkey: str, model_id: ModelId) -> str:
+    return os.path.join(
+        get_local_miner_dir(base_dir, hotkey),
+        "models" + "--" + model_id.namespace + "--" + model_id.name,
+    )
+
+
+def get_local_model_snapshot_dir(base_dir: str, hotkey: str, model_id: ModelId) -> str:
+    return os.path.join(
+        get_local_model_dir(base_dir, hotkey, model_id),
+        "snapshots",
+        model_id.commit,
+    )
+
+
+def get_hf_download_path(local_path: str, model_id: ModelId) -> str:
+    return os.path.join(
+        local_path,
+        "models" + "--" + model_id.namespace + "--" + model_id.name,
+        "snapshots",
+        model_id.commit,
+    )
+
+
+def get_newest_datetime_under_path(path: str) -> datetime.datetime:
+    newest_filetime = sys.maxsize
+
+    # Check to see if any file at any level was modified more recently than the current one.
+    for cur_path, dirnames, filenames in os.walk(path):
+        for filename in filenames:
+            path = os.path.join(cur_path, filename)
+            try:
+                mod_time = os.stat(path).st_mtime
+                if mod_time < newest_filetime:
+                    newest_filetime = mod_time
+            except:
+                pass
+
+    if newest_filetime == sys.maxsize:
+        return datetime.datetime.max
+
+    return datetime.datetime.fromtimestamp(newest_filetime)
+
+
+def remove_dir_out_of_grace_by_datetime(path: str, grace_period_seconds: int, last_modified: datetime.datetime) -> bool:
+    """Removes a dir if the last modified time is out of grace period secs. Returns if it was deleted."""
+    grace = datetime.timedelta(seconds=grace_period_seconds)
+
+    if last_modified < datetime.datetime.now() - grace:
+        shutil.rmtree(path=path, ignore_errors=True)
+        return True
+
+    return False
+
+def remove_dir_out_of_grace(path: str, grace_period_seconds: int) -> bool:
+    """Removes a dir if the last modified time is out of grace period secs. Returns if it was deleted."""
+    last_modified = get_newest_datetime_under_path(path)
+    return remove_dir_out_of_grace_by_datetime(path, grace_period_seconds, last_modified)
+
+
+def realize_symlinks_in_directory(path: str) -> int:
+    """Realizes all symlinks in the given directory, moving the linked file to the location. Returns count removed."""
+    realized_symlinks = 0
+
+    for cur_path, dirnames, filenames in os.walk(path):
+        for filename in filenames:
+            path = os.path.abspath(os.path.join(cur_path, filename))
+            # Get path resolving symlinks if encountered
+            real_path = os.path.realpath(path)
+            # If different then move
+            if path != real_path:
+                realized_symlinks += 1
+                shutil.move(real_path, path)
+
+    return realized_symlinks
+
+
+def get_hash_of_file(path: str) -> str:
+    blocksize = 64 * 1024
+    file_hash = hashlib.sha256()
+    with open(path, "rb") as fp:
+        while True:
+            data = fp.read(blocksize)
+            if not data:
+                break
+            file_hash.update(data)
+    return base64.b64encode(file_hash.digest()).decode("utf-8")
+
+
+def get_hash_of_directory(path: str) -> str:
+    dir_hash = hashlib.sha256()
+
+    # Recursively walk everything under the directory for files.
+    for cur_path, dirnames, filenames in os.walk(path):
+        # Ensure we walk future directories in a consistent order.
+        dirnames.sort()
+        # Ensure we walk files in a consistent order.
+        for filename in sorted(filenames):
+            path = os.path.join(cur_path, filename)
+            file_hash = get_hash_of_file(path)
+            dir_hash.update(file_hash.encode())
+
+    return base64.b64encode(dir_hash.digest()).decode("utf-8")

model/storage/eval_leaderboard.py

@@ -0,0 +1,185 @@
+from sqlalchemy import create_engine, Column, Integer, Float, String, DateTime, ForeignKey
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import Session, sessionmaker, relationship
+from sqlalchemy.exc import OperationalError
+from contextlib import contextmanager
+from datetime import datetime
+import bittensor as bt
+from typing import Optional, Dict, List
+import time
+from vali_api.config import DBHOST, DBNAME, DBUSER, DBPASS
+
+Base = declarative_base()
+
+# Global variables for engine and Session
+_engine: Optional[object] = None
+Session: Optional[sessionmaker] = None
+
+def init_database():
+    """Initialize the database connection and create tables."""
+    global _engine, Session
+    
+    if _engine is not None:
+        bt.logging.warning("Database already initialized")
+        return
+
+    try:
+        connection_string = f'mysql://{DBUSER}:{DBPASS}@{DBHOST}/{DBNAME}'
+        _engine = create_engine(connection_string)
+        Session = sessionmaker(bind=_engine)
+        Base.metadata.create_all(_engine)
+        bt.logging.info("Database initialized successfully")
+    except Exception as e:
+        bt.logging.error(f"Failed to initialize database: {str(e)}")
+        raise
+
+def get_session() -> Session:
+    """Get a database session."""
+    if Session is None:
+        raise RuntimeError("Database not initialized. Call init_database() first.")
+    return Session()
+
+class EvaluationModel(Base):
+    __tablename__ = 'sn21_evals_test'
+
+    eval_id = Column(Integer, primary_key=True)
+    miner_hotkey = Column(String(255))
+    miner_uid = Column(Integer)
+    model_name = Column(String(255))
+    model_type = Column(String(255))
+    eval_date = Column(DateTime)
+    competition_id = Column(String(10))
+    
+    results = relationship("EvaluationResult", back_populates="evaluation")
+
+class EvaluationResult(Base):
+    __tablename__ = 'sn21_eval_results_test'
+
+    eval_result_id = Column(Integer, primary_key=True)
+    eval_id = Column(Integer, ForeignKey('sn21_evals_test.eval_id'))
+    task = Column(String(255))
+    result_name = Column(String(255))
+    result = Column(Float)
+    competition_id = Column(String(10))
+    deleted_at = Column(DateTime)
+
+    evaluation = relationship("EvaluationModel", back_populates="results")
+
+class EvalLeaderboardManager:
+    def __init__(self, max_retries=3, retry_delay=1):
+        if Session is None:
+            raise RuntimeError("Database not initialized. Call init_database() first.")
+            
+        self.session = get_session()
+        self.max_retries = max_retries
+        self.retry_delay = retry_delay
+
+    @contextmanager
+    def session_scope(self):
+        """Provide a transactional scope around a series of operations."""
+        session = get_session()
+        try:
+            yield session
+            session.commit()
+        except Exception as e:
+            session.rollback()
+            raise
+        finally:
+            session.close()
+
+    def execute_with_retry(self, operation, *args, **kwargs):
+        """Execute database operation with retry logic."""
+        for attempt in range(self.max_retries):
+            try:
+                return operation(*args, **kwargs)
+            except OperationalError as e:
+                if attempt < self.max_retries - 1:
+                    bt.logging.warning(f"Database error. Attempt {attempt + 1}/{self.max_retries}. Retrying...")
+                    time.sleep(self.retry_delay)
+                else:
+                    raise
+            except Exception as e:
+                bt.logging.error(f"Error executing operation: {str(e)}")
+                raise
+
+    
+    def get_metrics_timeseries(self) -> Dict[str, List[Dict]]:
+        """
+        Get time series data for all turn metrics.
+        Returns data in format {metric_name: [{date: "", models: [{modelType: "", modelName: "", score: 123}, ...]}, ...]}
+        Groups models by model_type and includes model names in the data.
+        """
+        def _get_timeseries():
+            with self.session_scope() as session:
+                evals = session.query(EvaluationModel).order_by(EvaluationModel.eval_date).all()
+            
+                metrics = set()
+                # Get all non-deleted results that aren't the excluded metric
+                sample_results = session.query(EvaluationResult).filter(
+                    EvaluationResult.result_name != "exact_match_stderr,flexible-extract",
+                    EvaluationResult.deleted_at.is_(None)  # SQLAlchemy's proper way to check for NULL
+                ).all()
+                for result in sample_results:
+                    metric_name = result.result_name
+                    metrics.add(metric_name)
+                timeseries_data = {metric: {} for metric in metrics}
+            
+                for eval_ in evals:
+                    date_str = eval_.eval_date.strftime('%Y-%m-%d')
+                    model_type = eval_.model_type
+                    model_name = eval_.model_name
+                    competition_id = eval_.competition_id
+                
+                    # Only process non-deleted results
+                    for result in [r for r in eval_.results if r.deleted_at is None]:
+                        metric_name = result.result_name
+                        if metric_name in metrics:
+                            if date_str not in timeseries_data[metric_name]:
+                                timeseries_data[metric_name][date_str] = {
+                                    'date': date_str,
+                                    'models': []
+                                }
+                            
+                            # Check if we already have an entry for this model_type
+                            existing_model = next(
+                                (m for m in timeseries_data[metric_name][date_str]['models'] 
+                                 if m['modelType'] == model_type),
+                                None
+                            )
+                            
+                            if existing_model:
+                                # Update existing entry
+                                existing_model['score'] = result.result
+                                if model_name not in existing_model['modelName']:
+                                    existing_model['modelName'] = model_name
+                            else:
+                                # Add new entry
+                                timeseries_data[metric_name][date_str]['models'].append({
+                                    'modelType': model_type,
+                                    'modelName': model_name,
+                                    'score': result.result,
+                                    'competition_id': competition_id,
+                                    'task': result.task
+                                })
+            
+                return {
+                    metric: [
+                        data_point for data_point in sorted(data.values(), key=lambda x: x['date'])
+                        if data_point['models']  # Filter out entries with empty models list
+                    ]
+                    for metric, data in timeseries_data.items()
+                }
+        
+        try:
+            return self.execute_with_retry(_get_timeseries)
+        except Exception as e:
+            bt.logging.error(f"Failed to get metrics timeseries: {str(e)}")
+            return {}
+
+
+    def close(self):
+        """Safely close the session."""
+        try:
+            self.session.close()
+        except:
+            pass

model/storage/hugging_face/hugging_face_model_store.py

@@ -0,0 +1,175 @@
+import os
+from omegaconf import OmegaConf
+from huggingface_hub import HfApi
+from model.data import Model, ModelId
+from model.storage.disk import utils
+from constants import CompetitionParameters, MAX_HUGGING_FACE_BYTES
+
+from model.storage.remote_model_store import RemoteModelStore
+from huggingface_hub import HfApi, file_exists
+from collections import defaultdict
+
+
+MODEL_FILE_PT = "meta_model_{epoch}.pt"
+ADAPTER_FILE_PT = "adapter_{epoch}.pt"
+CONFIG_FILE = "training_config.yml"
+CONFIG_FILE_MOSHI = "config.yaml"
+README_FILE = "README.md"
+HOTKEY_FILE = "hotkey.txt"
+
+### MOSHI ###
+LM_FILE_PT_MOSHI = "model.safetensors"
+MIMI_FILE_PT_MOSHI = "tokenizer-e351c8d8-checkpoint125.safetensors"
+TOKENIZER_FILE_MOSHI = "tokenizer_spm_32k_3.model"
+
+
+def check_config(ckpt_dir):
+    config_file = os.path.join(ckpt_dir, CONFIG_FILE)
+    cfg = OmegaConf.load(config_file)
+    if cfg.model.use_clip:
+        raise ValueError("Cannot upload checkpoints with CLIP embeddings")
+
+
+def get_required_files(epoch: int, model_type: str):
+    if model_type == "o1":
+        return [
+            MODEL_FILE_PT.format(epoch=epoch),
+            CONFIG_FILE,
+        ]
+    elif model_type == "v1":
+        return [
+            LM_FILE_PT_MOSHI,
+            MIMI_FILE_PT_MOSHI,
+            TOKENIZER_FILE_MOSHI,
+            CONFIG_FILE_MOSHI
+        ]
+
+
+def export_readme(ckpt_dir: str):
+    readme_file = os.path.join(ckpt_dir, README_FILE)
+    with open(readme_file, "w") as f:
+        f.write(
+            f"""---
+license: mit
+tags:
+- any-to-any
+- omega
+- omegalabs
+- bittensor
+- agi
+---
+
+This is an Any-to-Any model checkpoint for the OMEGA Labs x Bittensor Any-to-Any subnet.
+
+Check out the [git repo](https://github.com/omegalabsinc/omegalabs-anytoany-bittensor) and find OMEGA on X: [@omegalabsai](https://x.com/omegalabsai).
+"""
+        )
+
+
+def export_hotkey(ckpt_dir: str, hotkey: str):
+    hotkey_file = os.path.join(ckpt_dir, HOTKEY_FILE)
+    with open(hotkey_file, "w") as f:
+        f.write(hotkey)
+
+
+class HuggingFaceModelStore(RemoteModelStore):
+    """Hugging Face based implementation for storing and retrieving a model."""
+
+    @classmethod
+    def assert_access_token_exists(cls) -> str:
+        """Asserts that the access token exists."""
+        if not os.getenv("HF_ACCESS_TOKEN"):
+            raise ValueError("No Hugging Face access token found to write to the hub.")
+        return os.getenv("HF_ACCESS_TOKEN")
+
+
+    async def upload_model(
+        self, model: Model, 
+        competition_parameters: CompetitionParameters,
+        hotkey: str
+    ) -> ModelId:
+        """Uploads a trained model to Hugging Face."""
+        token = HuggingFaceModelStore.assert_access_token_exists()
+        api = HfApi(token=token)
+        export_readme(model.local_repo_dir)
+        export_hotkey(model.local_repo_dir, hotkey)
+        hf_repo_id = model.id.namespace + "/" + model.id.name
+        api.create_repo(
+            repo_id=hf_repo_id,
+            exist_ok=True,
+            private=True,
+        )
+        commit_info = api.upload_folder(repo_id=hf_repo_id, folder_path=model.local_repo_dir, use_auth_token=True)
+        
+        print(f"Successfully uploaded model repository '{model.local_repo_dir}' to {hf_repo_id}")
+        
+        model_id_with_commit = ModelId(
+            namespace=model.id.namespace,
+            name=model.id.name,
+            epoch=model.id.epoch,
+            hash=model.id.hash,
+            commit=commit_info.oid,
+            competition_id=model.id.competition_id,
+        )
+        
+        return model_id_with_commit
+        # # TODO consider skipping the redownload if a hash is already provided.
+        # # To get the hash we need to redownload it at a local tmp directory after which it can be deleted.
+        # with tempfile.TemporaryDirectory() as temp_dir:
+        #     model_with_hash = await self.download_model(
+        #         model_id_with_commit, temp_dir, competition_parameters
+        #     )
+        #     # Return a ModelId with both the correct commit and hash.
+        #     return model_with_hash.id
+
+    async def download_model(
+        self,
+        model_id: ModelId,
+        local_path: str,
+        model_parameters: CompetitionParameters,
+    ) -> Model:
+        """Retrieves a trained model from Hugging Face."""
+        if not model_id.commit:
+            raise ValueError("No Hugging Face commit id found to read from the hub.")
+
+        repo_id = model_id.namespace + "/" + model_id.name
+
+        # Check ModelInfo for the size of model.safetensors file before downloading.
+        try:
+            token = HuggingFaceModelStore.assert_access_token_exists()
+        except:
+            token = None
+        api = HfApi(token=token)
+        model_info = api.model_info(
+            repo_id=repo_id, revision=model_id.commit, timeout=10, files_metadata=True
+        )
+        size = sum(repo_file.size for repo_file in model_info.siblings)
+        if size > MAX_HUGGING_FACE_BYTES:
+            raise ValueError(
+                f"Hugging Face repo over maximum size limit. Size {size}. Limit {MAX_HUGGING_FACE_BYTES}."
+            )
+
+        api.hf_hub_download(
+            repo_id=repo_id,
+            revision=model_id.commit,
+            filename="checkpoint.safetensors",
+            cache_dir=local_path,
+        )
+
+        # Get the directory the model was stored to.
+        model_dir = utils.get_hf_download_path(local_path, model_id)
+
+        # Realize all symlinks in that directory since Transformers library does not support avoiding symlinks.
+        utils.realize_symlinks_in_directory(model_dir)
+
+        # Compute the hash of the downloaded model.
+        model_hash = utils.get_hash_of_directory(model_dir)
+        model_id_with_hash = ModelId(
+            namespace=model_id.namespace,
+            name=model_id.name,
+            commit=model_id.commit,
+            hash=model_hash,
+            competition_id=model_id.competition_id,
+        )
+
+        return Model(id=model_id_with_hash, ckpt=model_dir)

model/storage/local_model_store.py

@@ -0,0 +1,30 @@
+import abc
+from typing import Dict
+from model.data import Model, ModelId
+from constants import CompetitionParameters
+
+
+class LocalModelStore(abc.ABC):
+    """An abstract base class for storing and retrieving a pre trained model locally."""
+
+    @abc.abstractmethod
+    def store_model(self, hotkey: str, model: Model) -> ModelId:
+        """Stores a trained model in the appropriate location based on implementation."""
+        pass
+
+    @abc.abstractmethod
+    def get_path(self, hotkey: str) -> str:
+        """Returns the path to the appropriate location based on implementation."""
+        pass
+
+    @abc.abstractmethod
+    def retrieve_model(self, hotkey: str, model_id: ModelId, parameters: CompetitionParameters) -> Model:
+        """Retrieves a trained model from the appropriate location based on implementation."""
+        pass
+
+    @abc.abstractmethod
+    def delete_unreferenced_models(
+        self, valid_models_by_hotkey: Dict[str, ModelId], grace_period_seconds: int
+    ):
+        """Check across all of local storage and delete unreferenced models out of grace period."""
+        pass

model/storage/model_metadata_store.py

@@ -0,0 +1,17 @@
+import abc
+from typing import Optional
+from model.data import ModelId, ModelMetadata
+
+
+class ModelMetadataStore(abc.ABC):
+    """An abstract base class for storing and retrieving model metadata."""
+
+    @abc.abstractmethod
+    async def store_model_metadata(self, hotkey: str, model_id: ModelId):
+        """Stores model metadata on this subnet for a specific miner."""
+        pass
+
+    @abc.abstractmethod
+    async def retrieve_model_metadata(self, hotkey: str) -> Optional[ModelMetadata]:
+        """Retrieves model metadata + block information on this subnet for specific miner, if present"""
+        pass

model/storage/mysql_model_queue.py

@@ -0,0 +1,847 @@
+from sqlalchemy import create_engine, Column, Integer, Float, String, DateTime, Boolean, JSON, func, desc, exists, ForeignKey, or_, and_, case, text
+from sqlalchemy.exc import OperationalError, SQLAlchemyError
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.dialects.mysql import JSON as MySQLJSON
+from sqlalchemy.orm import Session, sessionmaker, aliased, relationship
+from contextlib import contextmanager
+from collections import defaultdict
+
+import time
+import json
+
+from datetime import datetime, timedelta, timezone
+import bittensor as bt
+from typing import Optional
+
+from model.data import ModelId
+from vali_api.config import DBHOST, DBNAME, DBUSER, DBPASS, IS_PROD
+
+Base = declarative_base()
+
+# Global variables for engine and Session
+_engine: Optional[object] = None
+Session: Optional[sessionmaker] = None
+
+def init_database():
+    """
+    Initialize the database connection and create tables.
+    Must be called before using any database operations.
+    """
+    global _engine, Session
+    
+    if _engine is not None:
+        bt.logging.warning("Database already initialized")
+        return
+
+    # Try different MySQL drivers in order of preference
+    drivers_to_try = [
+        ('mysql', 'mysqlclient (MySQLdb)'),
+        ('mysql+pymysql', 'PyMySQL')
+    ]
+    
+    for driver, driver_name in drivers_to_try:
+        try:
+            connection_string = f'{driver}://{DBUSER}:{DBPASS}@{DBHOST}/{DBNAME}'
+            bt.logging.info(f"Attempting database connection with {driver_name}")
+            
+            _engine = create_engine(connection_string)
+            Session = sessionmaker(bind=_engine)
+            
+            # Test the connection
+            with Session() as session:
+                session.execute(text('SELECT 1'))
+            
+            # Create all tables
+            Base.metadata.create_all(_engine)
+            bt.logging.info(f"Database initialized successfully with {driver_name}")
+            return
+            
+        except ImportError as e:
+            bt.logging.warning(f"Driver {driver_name} not available: {e}")
+            continue
+        except Exception as e:
+            bt.logging.error(f"Failed to connect with {driver_name}: {e}")
+            if driver == drivers_to_try[-1][0]:  # Last driver in list
+                raise
+            continue
+    
+    raise RuntimeError("Failed to initialize database with any available MySQL driver")
+
+def get_session() -> Session:
+    """
+    Get a database session. Raises exception if database not initialized.
+    """
+    if Session is None:
+        raise RuntimeError("Database not initialized. Call init_database() first.")
+    return Session()
+
+def get_table_name(base_name: str) -> str:
+    """Helper function to get the correct table name with suffix if not in production."""
+    return f"{base_name}{'_test' if not IS_PROD else ''}"
+
+class ModelQueue(Base):
+    __tablename__ = get_table_name('sn21_model_queue')
+
+    hotkey = Column(String(255), primary_key=True)
+    uid = Column(String(255), primary_key=True, index=True)
+    block = Column(Integer, index=True)
+    competition_id = Column(String(255), index=True)
+    model_metadata = Column(JSON)
+    is_new = Column(Boolean, default=True)
+    is_being_scored = Column(Boolean, default=False)
+    is_being_scored_by = Column(String(255), default=None)
+    scoring_updated_at = Column(DateTime, default=None)
+    updated_at = Column(DateTime, default=datetime.utcnow)
+
+    # Relationship to use dynamic table name (lambda function)
+    scores = relationship(
+        "ScoreHistory",
+        back_populates="model",
+        foreign_keys="[ScoreHistory.hotkey, ScoreHistory.uid]",
+        primaryjoin=lambda: and_(
+            ModelQueue.hotkey == ScoreHistory.hotkey,
+            ModelQueue.uid == ScoreHistory.uid
+        )
+    )
+
+    def __repr__(self):
+        return f"<ModelQueue(hotkey='{self.hotkey}', uid='{self.uid}', competition_id='{self.competition_id}', is_new={self.is_new})>"
+
+class ScoreHistory(Base):
+    __tablename__ = get_table_name('sn21_score_history')
+
+    id = Column(Integer, primary_key=True)
+    hotkey = Column(String(255), ForeignKey(f"{get_table_name('sn21_model_queue')}.hotkey", ondelete='SET NULL'), index=True, nullable=True)
+    uid = Column(String(255), ForeignKey(f"{get_table_name('sn21_model_queue')}.uid", ondelete='SET NULL'), index=True, nullable=True)
+    competition_id = Column(String(255), index=True)
+    model_metadata = Column(JSON)
+    score = Column(Float)
+    scored_at = Column(DateTime, default=datetime.utcnow)
+    block = Column(Integer)
+    model_hash = Column(String(255))
+    scorer_hotkey = Column(String(255), index=True)
+    is_archived = Column(Boolean, default=False)
+    metric_scores = Column(MySQLJSON, nullable=True)
+    wandb_run_id = Column(String(255), nullable=True)
+    wandb_run_url = Column(String(512), nullable=True)
+    # Relationship to ModelQueue using dynamic table name (lambda function)
+    model = relationship(
+        "ModelQueue",
+        back_populates="scores",
+        foreign_keys=[hotkey, uid],
+        primaryjoin=lambda: and_(
+            ModelQueue.hotkey == ScoreHistory.hotkey,
+            ModelQueue.uid == ScoreHistory.uid
+        )
+    )
+
+    def __repr__(self):
+        return f"<ScoreHistory(hotkey='{self.hotkey}', uid='{self.uid}', score={self.score}, scored_at={self.scored_at}, model_metadata={self.model_metadata} is_archived={self.is_archived})>"
+
+class ModelIdEncoder(json.JSONEncoder):
+    def default(self, obj):
+        if isinstance(obj, ModelId):
+            return {
+                'namespace': obj.namespace,
+                'name': obj.name,
+                'epoch': obj.epoch,
+                'commit': obj.commit,
+                'hash': obj.hash,
+                'competition_id': obj.competition_id
+            }
+        return super().default(obj)
+
+class ModelQueueManager:
+    def __init__(self, max_scores_per_model=5, rescore_interval_hours=24, max_retries=3, retry_delay=1):
+        if Session is None:
+            raise RuntimeError("Database not initialized. Call init_database() first.")
+
+        self.session = get_session()
+        self.max_scores_per_model = max_scores_per_model
+        self.rescore_interval = timedelta(hours=rescore_interval_hours)
+        self.max_retries = max_retries
+        self.retry_delay = retry_delay
+
+    @contextmanager
+    def session_scope(self):
+        """Provide a transactional scope around a series of operations."""
+        session = get_session()
+        try:
+            yield session
+            session.commit()
+        except Exception as e:
+            session.rollback()
+            raise
+        finally:
+            session.close()
+
+    def reset_session(self):
+        """Reset the session in case of connection issues."""
+        try:
+            self.session.close()
+        except:
+            pass
+        try:
+            self.session = get_session()
+        except RuntimeError as e:
+            bt.logging.error(f"Failed to reset session: {str(e)}")
+            raise
+
+    def execute_with_retry(self, operation, *args, **kwargs):
+        """Execute an operation with retry logic."""
+        for attempt in range(self.max_retries):
+            try:
+                return operation(*args, **kwargs)
+            except OperationalError as e:
+                if "Lost connection" in str(e) and attempt < self.max_retries - 1:
+                    bt.logging.warning(f"Lost connection to MySQL. Attempt {attempt + 1}/{self.max_retries}. Retrying...")
+                    self.reset_session()
+                    time.sleep(self.retry_delay)
+                else:
+                    raise
+            except SQLAlchemyError as e:
+                if attempt < self.max_retries - 1:
+                    bt.logging.warning(f"Database error. Attempt {attempt + 1}/{self.max_retries}. Retrying...")
+                    self.reset_session()
+                    time.sleep(self.retry_delay)
+                else:
+                    raise
+
+    def store_updated_model(self, uid, hotkey, model_metadata, updated):
+        """
+        Store or update model metadata with retry logic.
+        
+        Args:
+            uid (str): Model UID
+            hotkey (str): Model hotkey
+            model_metadata: Model metadata object
+            updated (bool): Whether this is an update
+            
+        Returns:
+            bool: Success status
+        """
+        def _store_model():
+            with self.session_scope() as session:
+                try:
+                    # Query existing model with lock
+                    existing_model = session.query(ModelQueue).filter_by(
+                        hotkey=hotkey, 
+                        uid=uid
+                    ).with_for_update().first()
+
+                    # Serialize metadata
+                    serialized_metadata = json.dumps(model_metadata.__dict__, cls=ModelIdEncoder)
+
+                    if existing_model:
+                        if existing_model.model_metadata != serialized_metadata or existing_model.block != model_metadata.block:
+                            bt.logging.debug(f"Updating existing model metadata for UID={uid}, Hotkey={hotkey}. Old metadata: {existing_model.model_metadata}, New metadata: {serialized_metadata}")
+                            existing_model.model_metadata = serialized_metadata
+                            existing_model.is_new = True
+                            existing_model.block = model_metadata.block
+                            existing_model.updated_at = datetime.utcnow()
+                    else:
+                        # Create new model entry
+                        new_model = ModelQueue(
+                            hotkey=hotkey,
+                            uid=uid,
+                            competition_id=model_metadata.id.competition_id,
+                            model_metadata=serialized_metadata,
+                            is_new=True,
+                            block=model_metadata.block
+                        )
+                        session.add(new_model)
+                        bt.logging.debug(f"Stored new model for UID={uid}, Hotkey={hotkey} in database. Is new = {updated}")
+
+                    return True
+
+                except Exception as e:
+                    bt.logging.error(f"Error in _store_model: {str(e)}")
+                    bt.logging.error(f"Model metadata: {model_metadata}")
+                    raise
+
+        try:
+            return self.execute_with_retry(_store_model)
+        except Exception as e:
+            bt.logging.error(f"Failed to store model after {self.max_retries} attempts: {str(e)}")
+            return False
+
+    def get_next_model_to_score(self, competition_id: str):
+        """
+        Get next model to score with retry logic.
+        
+        The updated prioritization logic ensures:
+        1. New models (highest priority)
+        2. Models never scored with non-zero scores
+        3a. High-scoring models not scored for over a week
+        3b. Models not scored for more than 7 days (safety net for winning models)
+        4. Models eligible by standard criteria (not scored in 5 days or < 5 scores)
+        5. Everything else (lowest priority)
+        
+        Zero-scored models that are frequently scored are downgraded in priority
+        to prevent them from consuming too many resources.
+        """
+        def _get_next_model():
+            with self.session_scope() as session:
+                try:
+                    now = datetime.utcnow()
+
+                    # ---- START: Query to find overall highest score ----
+                    overall_max_score_value = session.query(func.max(ScoreHistory.score)).filter(
+                        ScoreHistory.competition_id == competition_id,
+                        ScoreHistory.is_archived == False,
+                        ScoreHistory.score > 0  # Consider only positive scores as relevant for "highest"
+                    ).scalar()
+
+                    if overall_max_score_value is not None:
+                        bt.logging.info(f"Overall highest positive score in competition '{competition_id}' is: {overall_max_score_value:.4f}")
+                    else:
+                        bt.logging.info(f"No positive scores found for competition '{competition_id}' to determine an overall highest score.")
+                    # ---- END: Query ----
+                    
+                    # Get latest score timestamp and count for each model
+                    score_subquery = session.query(
+                        ScoreHistory.hotkey,
+                        ScoreHistory.uid,
+                        func.count(ScoreHistory.id).label('score_count'),
+                        func.max(ScoreHistory.scored_at).label('latest_scored_at'),  # Get the latest score timestamp
+                        func.max(ScoreHistory.score).label('max_score')  # Get the maximum score
+                    ).filter(
+                        ScoreHistory.is_archived == False,
+                        ScoreHistory.competition_id == competition_id,
+                        ScoreHistory.score > 0  # Only consider non-zero scores
+                    ).group_by(
+                        ScoreHistory.hotkey, 
+                        ScoreHistory.uid
+                    ).subquery()
+
+                    # Also track all scores (including zeros) for high-frequency zero score detection
+                    all_scores_subquery = session.query(
+                        ScoreHistory.hotkey,
+                        ScoreHistory.uid,
+                        func.count(ScoreHistory.id).label('all_score_count'),
+                        func.max(ScoreHistory.scored_at).label('latest_all_scored_at'),
+                        func.sum(case((ScoreHistory.score > 0, 1), else_=0)).label('non_zero_count')
+                    ).filter(
+                        ScoreHistory.is_archived == False,
+                        ScoreHistory.competition_id == competition_id
+                    ).group_by(
+                        ScoreHistory.hotkey, 
+                        ScoreHistory.uid
+                    ).subquery()
+
+                    five_days_ago = now - timedelta(days=5)
+                    weekly_rescore_threshold_time = now - timedelta(days=7)  # Define a 7-day threshold
+                    
+                    # Check if we have new models before proceeding
+                    have_new_models = session.query(ModelQueue).filter(
+                        ModelQueue.is_being_scored == False,
+                        ModelQueue.competition_id == competition_id,
+                        ModelQueue.is_new == True
+                    ).first() is not None
+                    
+                    # Check if we have never-scored models
+                    never_scored_count = session.query(func.count(ModelQueue.uid)).filter(
+                        ModelQueue.is_being_scored == False,
+                        ModelQueue.competition_id == competition_id,
+                        ~exists().where(
+                            and_(
+                                ScoreHistory.hotkey == ModelQueue.hotkey,
+                                ScoreHistory.uid == ModelQueue.uid,
+                                ScoreHistory.score > 0
+                            )
+                        )
+                    ).scalar()
+                    
+                    # If no new models and no never-scored models, prioritize high scoring models not scored recently
+                    if not have_new_models:
+                        # ---- START: Modified logic for dynamic high-score threshold ----
+                        if overall_max_score_value is not None and overall_max_score_value > 0: # Ensure we have a valid max score
+                            dynamic_high_score_threshold = overall_max_score_value * 0.97
+                            bt.logging.info(f"Using dynamic high-score threshold for competition '{competition_id}': >= {dynamic_high_score_threshold:.4f} (based on overall max of {overall_max_score_value:.4f})")
+
+                            # First try to get a high-scoring model not scored in over a week
+                            top_model = session.query(ModelQueue).join(
+                                score_subquery,
+                                and_(
+                                    ModelQueue.hotkey == score_subquery.c.hotkey,
+                                    ModelQueue.uid == score_subquery.c.uid
+                                )
+                            ).filter(
+                                ModelQueue.is_being_scored == False,
+                                ModelQueue.competition_id == competition_id,
+                                score_subquery.c.latest_scored_at < weekly_rescore_threshold_time,
+                                score_subquery.c.max_score >= dynamic_high_score_threshold  # Use dynamic threshold
+                            ).order_by(
+                                score_subquery.c.max_score.desc()  # Highest score first
+                            ).with_for_update().first()
+                            
+                            if top_model:
+                                # Create a dictionary with the model's attributes
+                                model_data = {
+                                    'hotkey': top_model.hotkey,
+                                    'uid': top_model.uid,
+                                    'block': top_model.block,
+                                    'competition_id': top_model.competition_id,
+                                    'model_metadata': top_model.model_metadata,
+                                    'is_new': top_model.is_new,
+                                    'is_being_scored': top_model.is_being_scored,
+                                    'is_being_scored_by': top_model.is_being_scored_by,
+                                    'scoring_updated_at': top_model.scoring_updated_at,
+                                    'updated_at': top_model.updated_at
+                                }
+                                bt.logging.debug(f"Found high-scoring model (dynamic threshold) to score: hotkey={model_data['hotkey']}, uid={model_data['uid']}")
+                                return model_data
+                        else:
+                            bt.logging.info(f"Skipping dynamic high-score prioritization for competition '{competition_id}' as no overall positive max score is available or it's zero.")
+                        # ---- END: Modified logic ----
+                    
+                    # Otherwise, use the standard prioritization logic with the zero-score detection
+                    next_model = session.query(ModelQueue).outerjoin(
+                        score_subquery, 
+                        and_(
+                            ModelQueue.hotkey == score_subquery.c.hotkey, 
+                            ModelQueue.uid == score_subquery.c.uid
+                        )
+                    ).outerjoin(
+                        all_scores_subquery,
+                        and_(
+                            ModelQueue.hotkey == all_scores_subquery.c.hotkey,
+                            ModelQueue.uid == all_scores_subquery.c.uid
+                        )
+                    ).filter(
+                        ModelQueue.is_being_scored == False,
+                        ModelQueue.competition_id == competition_id
+                    ).order_by(
+                        desc(ModelQueue.is_new),  # 1. Prioritize new models
+                        (score_subquery.c.score_count == None).desc(),  # 2. Prioritize models never scored (non-zero)
+                        case( # 3. Prioritize models not scored for more than 7 days (safety net)
+                            (and_(score_subquery.c.latest_scored_at != None, score_subquery.c.latest_scored_at < weekly_rescore_threshold_time), 0),
+                            else_=1
+                        ),
+                        # 4. Decrease priority for models with all zero scores and frequent scoring
+                        case(
+                            (and_(
+                                all_scores_subquery.c.all_score_count > 10,  # Has many scores
+                                all_scores_subquery.c.non_zero_count == 0,   # All scores are zero
+                                all_scores_subquery.c.latest_all_scored_at > five_days_ago  # Scored recently
+                            ), 1),
+                            else_=0
+                        ),
+                        case( # 5. Prioritize models eligible by standard criteria
+                            (or_(
+                                score_subquery.c.latest_scored_at == None,
+                                score_subquery.c.latest_scored_at <= five_days_ago,
+                                score_subquery.c.score_count < 5
+                            ), 0),
+                            else_=1
+                        ),
+                        func.rand()  # 6. Random tie-breaker
+                    ).with_for_update().first()
+
+                    if next_model:
+                        # Create a dictionary with the model's attributes
+                        model_data = {
+                            'hotkey': next_model.hotkey,
+                            'uid': next_model.uid,
+                            'block': next_model.block,
+                            'competition_id': next_model.competition_id,
+                            'model_metadata': next_model.model_metadata,
+                            'is_new': next_model.is_new,
+                            'is_being_scored': next_model.is_being_scored,
+                            'is_being_scored_by': next_model.is_being_scored_by,
+                            'scoring_updated_at': next_model.scoring_updated_at,
+                            'updated_at': next_model.updated_at
+                        }
+                        bt.logging.debug(f"Found next model to score: hotkey={model_data['hotkey']}, uid={model_data['uid']}")
+                        return model_data
+                    else:
+                        bt.logging.debug("No models available for scoring")
+                        return None
+
+                except Exception as e:
+                    bt.logging.error(f"Error in _get_next_model: {str(e)}")
+                    raise
+
+        try:
+            return self.execute_with_retry(_get_next_model)
+        except Exception as e:
+            bt.logging.error(f"Failed to get next model after {self.max_retries} attempts: {str(e)}")
+            return None
+
+    def mark_model_as_being_scored(self, model_hotkey, model_uid, scorer_hotkey):
+        """Mark model as being scored with retry logic."""
+        def _mark_model():
+            with self.session_scope() as session:
+                model = session.query(ModelQueue).filter_by(
+                    hotkey=model_hotkey, 
+                    uid=model_uid
+                ).with_for_update().first()
+                
+                if model and not model.is_being_scored:
+                    model.is_being_scored = True
+                    model.is_being_scored_by = scorer_hotkey
+                    model.scoring_updated_at = datetime.utcnow()
+                    return True
+                return False
+
+        try:
+            return self.execute_with_retry(_mark_model)
+        except Exception as e:
+            bt.logging.error(f"Failed to mark model as being scored after {self.max_retries} attempts: {str(e)}")
+            return False
+
+    def submit_score(self, model_hotkey, model_uid, scorer_hotkey, model_hash, score, metric_scores):
+        """Submit score with retry logic. Mark the model in queue as scored. Remove from queue."""
+        def _submit_score():
+            with self.session_scope() as session:
+                try:
+                    model = session.query(ModelQueue).filter_by(
+                        hotkey=model_hotkey, 
+                        uid=model_uid
+                    ).with_for_update().first()
+                    
+                    if not model:
+                        bt.logging.error(f"No model found for hotkey {model_hotkey} and uid {model_uid}")
+                        return False
+
+                    """
+                    # temporarily allow scoring from any hotkey
+                    new_score = ScoreHistory(
+                        hotkey=model_hotkey,
+                        uid=model_uid,
+                        competition_id=model.competition_id,
+                        score=score,
+                        block=model.block,
+                        model_hash=model_hash,
+                        scorer_hotkey=scorer_hotkey,
+                        model_metadata=model.model_metadata 
+                    )
+                    session.add(new_score)
+                    model.is_new = False
+                    model.is_being_scored = False
+                    model.is_being_scored_by = None
+                    model.scoring_updated_at = None
+                    model.updated_at = datetime.now(timezone.utc)
+                    bt.logging.info(f"Successfully submitted score for model {model_hotkey} by {scorer_hotkey}")
+                    return True
+                    """
+                    
+                    if model.is_being_scored and model.is_being_scored_by == scorer_hotkey:
+                        # Extract wandb fields from metric_scores if present
+                        wandb_run_id = None
+                        wandb_run_url = None
+                        if metric_scores and isinstance(metric_scores, dict):
+                            wandb_run_id = metric_scores.get('wandb_run_id')
+                            wandb_run_url = metric_scores.get('wandb_run_url')
+                        
+                        new_score = ScoreHistory(
+                            hotkey=model_hotkey,
+                            uid=model_uid,
+                            competition_id=model.competition_id,
+                            score=score,
+                            block=model.block,
+                            model_hash=model_hash,
+                            scorer_hotkey=scorer_hotkey,
+                            model_metadata=model.model_metadata,
+                            metric_scores=metric_scores,
+                            wandb_run_id=wandb_run_id,
+                            wandb_run_url=wandb_run_url
+                        )
+                        session.add(new_score)
+                        model.is_new = False
+                        model.is_being_scored = False
+                        model.is_being_scored_by = None
+                        model.scoring_updated_at = None
+                        model.updated_at = datetime.now(timezone.utc)
+                        bt.logging.info(f"Successfully submitted score for model {model_hotkey} by {scorer_hotkey}")
+                        return True
+                    else:
+                        bt.logging.error(f"Failed to submit score for model {model_hotkey} by {scorer_hotkey}. "
+                                    f"Model: {model}, is_being_scored: {model.is_being_scored}, "
+                                    f"is_being_scored_by: {model.is_being_scored_by}")
+                        return False
+
+                except Exception as e:
+                    bt.logging.error(f"Error in _submit_score: {str(e)}")
+                    raise
+
+        try:
+            return self.execute_with_retry(_submit_score)
+        except Exception as e:
+            bt.logging.error(f"Failed to submit score after {self.max_retries} attempts: {str(e)}")
+            return False
+
+    def reset_stale_scoring_tasks(self, max_scoring_time_minutes=15):
+        """Reset stale scoring tasks with retry logic."""
+        def _reset_stale_tasks():
+            with self.session_scope() as session:
+                try:
+                    stale_time = datetime.utcnow() - timedelta(minutes=max_scoring_time_minutes)
+                    stale_models = session.query(ModelQueue).filter(
+                        ModelQueue.is_being_scored == True,
+                        ModelQueue.scoring_updated_at < stale_time
+                    ).with_for_update().all()
+
+                    reset_count = 0
+                    for model in stale_models:
+                        model.is_being_scored = False
+                        model.is_being_scored_by = None
+                        model.scoring_updated_at = None
+                        reset_count += 1
+                        bt.logging.info(f"Reset scoring task for stale model: hotkey={model.hotkey}, uid={model.uid}")
+
+                    return reset_count
+
+                except Exception as e:
+                    bt.logging.error(f"Error in _reset_stale_tasks: {str(e)}")
+                    raise
+
+        try:
+            return self.execute_with_retry(_reset_stale_tasks)
+        except Exception as e:
+            bt.logging.error(f"Failed to reset stale tasks after {self.max_retries} attempts: {str(e)}")
+            return 0
+    
+    def get_recent_model_scores(self, scores_per_model):
+        """
+        Get recent scores for all models.
+        
+        Args:
+            scores_per_model (int): Number of recent scores to fetch per model
+            
+        Returns:
+            dict: Dictionary of model scores grouped by UID
+        """
+        def _get_recent_scores():
+            with self.session_scope() as session:
+                try:
+                    # First, create a subquery that ranks scores by timestamp for each model
+                    ranked_scores = (
+                        session.query(
+                            ScoreHistory,
+                            func.row_number().over(
+                                partition_by=(ScoreHistory.hotkey, ScoreHistory.uid),
+                                order_by=desc(ScoreHistory.scored_at)
+                            ).label('score_rank')
+                        )
+                        .filter(ScoreHistory.is_archived == False)
+                        .filter(ScoreHistory.score != 0)
+                        .subquery()
+                    )
+
+                    # Get the most recent scores for each model
+                    recent_scores = session.query(ranked_scores).filter(
+                        ranked_scores.c.score_rank <= scores_per_model
+                    ).subquery('recent_scores')
+
+                    # Join with ModelQueue to get additional model information
+                    results = session.query(
+                        ModelQueue.uid,
+                        ModelQueue.hotkey,
+                        ModelQueue.competition_id,
+                        ModelQueue.model_metadata,
+                        recent_scores.c.score,
+                        recent_scores.c.scored_at,
+                        recent_scores.c.block,
+                        recent_scores.c.model_hash,
+                        recent_scores.c.scorer_hotkey,
+                        recent_scores.c.score_rank
+                    ).outerjoin(
+                        recent_scores,
+                        and_(
+                            ModelQueue.hotkey == recent_scores.c.hotkey,
+                            ModelQueue.uid == recent_scores.c.uid,
+                        )
+                    ).order_by(
+                        ModelQueue.uid,
+                        ModelQueue.hotkey,
+                        recent_scores.c.scored_at.desc()
+                    ).all()
+
+                    scores_by_uid = defaultdict(lambda: defaultdict(list))
+                    
+                    for result in results:
+                        if result.score is not None:
+                            # Create a unique key for each hotkey+uid combination
+                            model_key = f"{result.hotkey}_{result.uid}"
+                            
+                            scores_by_uid[result.uid][model_key].append({
+                                'hotkey': result.hotkey,
+                                'competition_id': result.competition_id,
+                                'model_metadata': result.model_metadata,
+                                'score': result.score,
+                                'scored_at': result.scored_at.isoformat() if result.scored_at else None,
+                                'block': result.block,
+                                'model_hash': result.model_hash,
+                                'scorer_hotkey': result.scorer_hotkey,
+                                'rank': result.score_rank
+                            })
+                        else:
+                            # Handle models with no scores
+                            model_key = f"{result.hotkey}_{result.uid}"
+                            if not scores_by_uid[result.uid][model_key]:  # Only add if no scores exist
+                                scores_by_uid[result.uid][model_key].append({
+                                    'hotkey': result.hotkey,
+                                    'competition_id': None,
+                                    'model_metadata': result.model_metadata,
+                                    'score': None,
+                                    'scored_at': None,
+                                    'block': None,
+                                    'model_hash': None,
+                                    'scorer_hotkey': None,
+                                    'rank': None
+                                })
+
+                    # Convert defaultdict to regular dict for return
+                    return {
+                        uid: dict(models) 
+                        for uid, models in scores_by_uid.items()
+                    }
+
+                except Exception as e:
+                    bt.logging.error(f"Error in _get_recent_scores: {str(e)}")
+                    raise
+
+        try:
+            return self.execute_with_retry(_get_recent_scores)
+        except Exception as e:
+            bt.logging.error(f"Failed to get recent scores after {self.max_retries} attempts: {str(e)}")
+            return {}
+    
+    def get_all_model_scores(self):
+        """Get all model scores with retry logic."""
+        def _get_all_scores():
+            with self.session_scope() as session:
+                try:
+                    # First, get the latest score timestamps
+                    latest_scores = session.query(
+                        ScoreHistory.hotkey,
+                        ScoreHistory.uid,
+                        func.max(ScoreHistory.scored_at).label('latest_score_time')
+                    ).filter(
+                        ScoreHistory.is_archived == False
+                    ).group_by(
+                        ScoreHistory.hotkey, 
+                        ScoreHistory.uid
+                    ).subquery('latest_scores')
+
+                    # Get score details
+                    latest_score_details = session.query(
+                        ScoreHistory
+                    ).join(
+                        latest_scores,
+                        and_(
+                            ScoreHistory.hotkey == latest_scores.c.hotkey,
+                            ScoreHistory.uid == latest_scores.c.uid,
+                            ScoreHistory.scored_at == latest_scores.c.latest_score_time
+                        )
+                    ).subquery('latest_score_details')
+
+                    # Get final results
+                    results = session.query(
+                        ModelQueue.uid,
+                        ModelQueue.hotkey,
+                        ModelQueue.competition_id,
+                        latest_score_details.c.score,
+                        latest_score_details.c.scored_at,
+                        latest_score_details.c.block,
+                        latest_score_details.c.model_hash,
+                        latest_score_details.c.scorer_hotkey
+                    ).outerjoin(
+                        latest_score_details,
+                        and_(
+                            ModelQueue.hotkey == latest_score_details.c.hotkey,
+                            ModelQueue.uid == latest_score_details.c.uid
+                        )
+                    ).all()
+
+                    scores_by_uid = defaultdict(list)
+                    for result in results:
+                        if result.score is not None:
+                            scores_by_uid[result.uid].append({
+                                'hotkey': result.hotkey,
+                                'competition_id': result.competition_id,
+                                'score': result.score,
+                                'scored_at': result.scored_at.isoformat() if result.scored_at else None,
+                                'block': result.block,
+                                'model_hash': result.model_hash,
+                            })
+                        else:
+                            scores_by_uid[result.uid].append({
+                                'hotkey': result.hotkey,
+                                'competition_id': result.competition_id,
+                                'score': None,
+                                'scored_at': None,
+                                'block': None,
+                                'model_hash': None,
+                            })
+
+                    return dict(scores_by_uid)
+
+                except Exception as e:
+                    bt.logging.error(f"Error in _get_all_scores: {str(e)}")
+                    raise
+
+        try:
+            return self.execute_with_retry(_get_all_scores)
+        except Exception as e:
+            bt.logging.error(f"Failed to get all scores after {self.max_retries} attempts: {str(e)}")
+            return {}
+
+    def archive_scores_for_deregistered_models(self, registered_hotkey_uid_pairs):
+        """Archive deregistered models with retry logic."""
+        def _archive_scores():
+            with self.session_scope() as session:
+                try:
+                    all_models = session.query(
+                        ModelQueue.hotkey, 
+                        ModelQueue.uid
+                    ).with_for_update().all()
+
+                    deregistered_models = set(
+                        (model.hotkey, model.uid) for model in all_models
+                    ) - set(registered_hotkey_uid_pairs)
+
+                    for hotkey, uid in deregistered_models:
+                        # Mark scores as archived
+                        archive_result = session.query(ScoreHistory).filter_by(
+                            hotkey=hotkey,
+                            uid=uid,
+                            is_archived=False
+                        ).update(
+                            {"is_archived": True},
+                            synchronize_session=False
+                        )
+
+                        # Remove from ModelQueue
+                        delete_result = session.query(ModelQueue).filter_by(
+                            hotkey=hotkey,
+                            uid=uid
+                        ).delete(synchronize_session=False)
+
+                        bt.logging.debug(
+                            f"Processed deregistered model - Hotkey: {hotkey}, "
+                            f"UID: {uid}, Archived scores: {archive_result}, "
+                            f"Removed from queue: {delete_result}"
+                        )
+
+                    return len(deregistered_models)
+
+                except Exception as e:
+                    bt.logging.error(f"Error in _archive_scores: {str(e)}")
+                    raise
+
+        try:
+            result = self.execute_with_retry(_archive_scores)
+            print(f"Archived scores and removed {result} deregistered models from the queue.")
+            return result
+        except Exception as e:
+            bt.logging.error(f"Failed to archive scores after {self.max_retries} attempts: {str(e)}")
+            return 0
+
+    def close(self):
+        """Safely close the session."""
+        try:
+            self.session.close()
+        except:
+            pass

model/storage/remote_model_store.py

@@ -0,0 +1,17 @@
+import abc
+from model.data import Model, ModelId
+from constants import CompetitionParameters
+from typing import Optional
+
+class RemoteModelStore(abc.ABC):
+    """An abstract base class for storing and retrieving a pre trained model."""
+
+    @abc.abstractmethod
+    async def upload_model(self, model: Model, parameters: CompetitionParameters) -> ModelId:
+        """Uploads a trained model in the appropriate location based on implementation."""
+        pass
+
+    @abc.abstractmethod
+    async def download_model(self, model_id: ModelId, local_path: str, parameters: CompetitionParameters) -> Model:
+        """Retrieves a trained model from the appropriate location and stores at the given path."""
+        pass

model/storage/reputation_store.py

@@ -0,0 +1,244 @@
+from sqlalchemy import create_engine, Column, Integer, Float, String, DateTime, Boolean, JSON, func, desc, exists, ForeignKey, or_, and_, case
+from sqlalchemy.exc import OperationalError, SQLAlchemyError
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import Session, sessionmaker, aliased, relationship
+from contextlib import contextmanager
+from collections import defaultdict
+
+import time
+from datetime import datetime, timedelta, timezone
+import bittensor as bt
+from typing import Optional
+
+from vali_api.config import DBHOST, DBNAME, DBUSER, DBPASS, IS_PROD
+
+
+Base = declarative_base()
+
+# Global variables for engine and Session
+_engine: Optional[object] = None
+Session: Optional[sessionmaker] = None
+
+def init_database():
+    """
+    Initialize the database connection and create tables.
+    Must be called before using any database operations.
+    """
+    global _engine, Session
+    
+    if _engine is not None:
+        bt.logging.warning("Database already initialized")
+        return
+
+    try:
+        connection_string = f'mysql://{DBUSER}:{DBPASS}@{DBHOST}/{DBNAME}'
+        _engine = create_engine(connection_string)
+        Session = sessionmaker(bind=_engine)
+        
+        # Create all tables
+        Base.metadata.create_all(_engine)
+        bt.logging.info("Database initialized successfully")
+        
+    except Exception as e:
+        bt.logging.error(f"Failed to initialize database: {str(e)}")
+        raise
+
+def get_session() -> Session:
+    """
+    Get a database session. Raises exception if database not initialized.
+    """
+    if Session is None:
+        raise RuntimeError("Database not initialized. Call init_database() first.")
+    return Session()
+
+def get_table_name(base_name: str) -> str:
+    """Helper function to get the correct table name with suffix if not in production."""
+    return f"{base_name}{'_test' if not IS_PROD else ''}"
+
+class BaselineScore(Base):
+    __tablename__ = get_table_name('sn21_baseline_scores')
+    id = Column(Integer, primary_key=True)
+    competition_id = Column(String(255), nullable=False, index=True)
+    score = Column(Float, nullable=False)
+    created_at = Column(DateTime, default=datetime.utcnow, index=True)
+
+    def __repr__(self):
+        return f"<BaselineScore(competition_id='{self.competition_id}', score={self.score}, created_at='{self.created_at}')>"
+
+class MinerReputation(Base):
+    __tablename__ = get_table_name('sn21_miner_reputations')
+    hotkey = Column(String(255), primary_key=True)
+    reputation = Column(Float, default=0.5, nullable=False)
+    last_updated = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+
+    def __repr__(self):
+        return f"<MinerReputation(hotkey='{self.hotkey}', reputation={self.reputation})>"
+
+class ReputationHistory(Base):
+    __tablename__ = get_table_name('sn21_reputation_history')
+    id = Column(Integer, primary_key=True)
+    hotkey = Column(String(255), nullable=False, index=True)
+    timestamp = Column(DateTime, nullable=False, index=True)
+    reputation = Column(Float, nullable=False)
+
+    def __repr__(self):
+        return f"<ReputationHistory(hotkey='{self.hotkey}', timestamp='{self.timestamp}', reputation={self.reputation})>"
+
+class ReputationStore:
+    def __init__(self, max_retries=3, retry_delay=1):
+        # Ensure DB is initialized and get the sessionmaker
+        if Session is None:
+            raise RuntimeError("Database not initialized. Call init_database() first.")
+        self.max_retries = max_retries
+        self.retry_delay = retry_delay
+        self.session = get_session()
+
+    @contextmanager
+    def session_scope(self):
+        """Provide a transactional scope around a series of operations."""
+        session = get_session()
+        try:
+            yield session
+            session.commit()
+        except Exception as e:
+            session.rollback()
+            raise
+        finally:
+            session.close()
+
+    def reset_session(self):
+        """Reset the session in case of connection issues."""
+        try:
+            self.session.close()
+        except:
+            pass
+        try:
+            self.session = get_session()
+        except Exception as e:
+            bt.logging.error(f"Failed to reset session: {str(e)}")
+            raise
+
+    def execute_with_retry(self, operation, *args, **kwargs):
+        """Execute an operation with retry logic."""
+        for attempt in range(self.max_retries):
+            try:
+                return operation(*args, **kwargs)
+            except OperationalError as e:
+                if "Lost connection" in str(e) and attempt < self.max_retries - 1:
+                    bt.logging.warning(f"Lost connection to MySQL. Attempt {attempt + 1}/{self.max_retries}. Retrying...")
+                    self.reset_session()
+                    time.sleep(self.retry_delay)
+                else:
+                    raise
+            except SQLAlchemyError as e:
+                if attempt < self.max_retries - 1:
+                    bt.logging.warning(f"Database error. Attempt {attempt + 1}/{self.max_retries}. Retrying...")
+                    self.reset_session()
+                    time.sleep(self.retry_delay)
+                else:
+                    raise
+
+    def get_latest_baseline_score(self, competition_id):
+        def _get():
+            with self.session_scope() as session:
+                latest_baseline = (
+                    session.query(BaselineScore)
+                    .filter(BaselineScore.competition_id == competition_id)
+                    .order_by(BaselineScore.created_at.desc())
+                    .first()
+                )
+                if latest_baseline is None:
+                    return None
+                else:
+                    return {"competition_id": latest_baseline.competition_id, "score": latest_baseline.score, "created_at": latest_baseline.created_at}
+
+        return self.execute_with_retry(_get)
+
+    def get_all_reputations(self):
+        def _get():
+            with self.session_scope() as session:
+                records = session.query(MinerReputation).all()
+                return {
+                    record.hotkey: {
+                        "reputation": record.reputation,
+                        "last_updated": record.last_updated.isoformat() if record.last_updated else None
+                    }
+                    for record in records
+                }
+        return self.execute_with_retry(_get)
+
+    def get_reputation(self, hotkey):
+        def _get():
+            with self.session_scope() as session:
+                record = session.query(MinerReputation).filter(MinerReputation.hotkey == hotkey).first()
+                if not record:
+                    return None
+                return {
+                    "hotkey": record.hotkey,
+                    "reputation": record.reputation,
+                    "last_updated": record.last_updated.isoformat() if record.last_updated else None
+                }
+        return self.execute_with_retry(_get) 
+
+def main():
+    """
+    Main function to demonstrate the ReputationStore's three get methods.
+    """
+    try:
+        # Initialize the database
+        print("Initializing database...")
+        init_database()
+        print("Database initialized successfully!")
+        
+        # Create ReputationStore instance
+        print("\nCreating ReputationStore instance...")
+        reputation_store = ReputationStore()
+        print("ReputationStore created successfully!")
+        
+        # Test 1: Get latest baseline score
+        print("\n=== Testing get_latest_baseline_score ===")
+        competition_id = "v1"
+        baseline_score = reputation_store.get_latest_baseline_score(competition_id)
+        if baseline_score:
+            print(f"Latest baseline score for competition '{competition_id}':")
+            print(f"  Score: {baseline_score['score']}")
+            print(f"  Created at: {baseline_score['created_at']}")
+        else:
+            print(f"No baseline score found for competition '{competition_id}'")
+        
+        # Test 2: Get all reputations
+        print("\n=== Testing get_all_reputations ===")
+        all_reputations = reputation_store.get_all_reputations()
+        if all_reputations:
+            print(f"Found {len(all_reputations)} miner reputations:")
+            for hotkey, data in all_reputations.items():
+                print(f"  Hotkey: {hotkey}")
+                print(f"    Reputation: {data['reputation']}")
+                print(f"    Last Updated: {data['last_updated']}")
+                break
+        else:
+            print("No miner reputations found in database")
+        
+        # Test 3: Get specific reputation
+        print("\n=== Testing get_reputation ===")
+        test_hotkey = "test_hotkey_123"
+        reputation = reputation_store.get_reputation(test_hotkey)
+        if reputation:
+            print(f"Reputation for hotkey '{test_hotkey}':")
+            print(f"  Hotkey: {reputation['hotkey']}")
+            print(f"  Reputation: {reputation['reputation']}")
+            print(f"  Last Updated: {reputation['last_updated']}")
+        else:
+            print(f"No reputation found for hotkey '{test_hotkey}'")
+        
+        print("\n=== All tests completed successfully! ===")
+        
+    except Exception as e:
+        print(f"Error occurred: {str(e)}")
+        import traceback
+        traceback.print_exc()
+
+if __name__ == "__main__":
+    main()
+
+

requirements.txt

@@ -0,0 +1,13 @@
+transformers==4.48.3
+pydantic==2.11.4
+numpy==2.2.5
+torch==2.6.0
+torchaudio==2.6.0
+torchvision==0.21.0
+fastapi==0.115.12
+uvicorn==0.34.2
+librosa==0.11.0
+openai-whisper==20240930
+soundfile==0.13.1
+accelerate==0.26.0
+voxcpm

server.py

@@ -0,0 +1,211 @@
+import os
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, Field
+from transformers import AutoModelForCausalLM, AutoTokenizer
+import traceback
+import whisper
+import librosa
+import numpy as np
+import torch
+import uvicorn
+import base64
+import io
+from voxcpm import VoxCPM
+
+
+asr_model = whisper.load_model("models/wpt/wpt.pt")
+model_name = "models/Llama-3.2-1B-Instruct"
+tok = AutoTokenizer.from_pretrained(model_name)
+lm = AutoModelForCausalLM.from_pretrained(
+    model_name,
+    torch_dtype=torch.bfloat16,
+    device_map="cuda",
+).eval()
+
+tts = VoxCPM.from_pretrained(
+    "models/VoxCPM-0.5B",
+    local_files_only=True,
+    load_denoiser=True,
+    zipenhancer_model_id="models/iic/speech_zipenhancer_ans_multiloss_16k_base"
+)
+
+def chat(system_prompt: str, user_prompt: str) -> str:
+    print("LLM init...")
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_prompt},
+    ]
+    inputs = tok.apply_chat_template(
+        messages,
+        add_generation_prompt=True,
+        return_tensors="pt",
+        return_dict=True
+    )
+    input_ids = inputs["input_ids"].to(lm.device)
+    attention_mask = inputs["attention_mask"].to(lm.device)
+
+    with torch.inference_mode():
+        output_ids = lm.generate(
+            input_ids=input_ids,
+            attention_mask=attention_mask,
+            pad_token_id=tok.eos_token_id,
+            max_new_tokens=2048,
+            do_sample=True,
+            temperature=0.2,
+            repetition_penalty=1.1,
+            top_k=100,
+            top_p=0.95,
+        )
+
+    answer = tok.decode(
+        output_ids[0][input_ids.shape[-1]:],
+        skip_special_tokens=True,
+        clean_up_tokenization_spaces=True,
+    )
+    print("LLM answer done.")
+    return answer.strip()
+
+def gt(audio: np.ndarray, sr: int):
+    print("Starting ASR transcription...")
+    ss = audio.squeeze().astype(np.float32)
+    if sr != 16_000:
+        ss = librosa.resample(audio, orig_sr=sr, target_sr=16_000)
+
+    result = asr_model.transcribe(ss, fp16=False, language=None)
+    transcribed_text = result["text"].strip()
+    print(f"ASR done. Transcribed: '{transcribed_text}'")
+    return transcribed_text
+
+
+def sample(rr: str) -> str:
+    if rr.strip() == "":
+        rr = "Hello "
+
+    inputs = tok(rr, return_tensors="pt").to(lm.device)
+
+    with torch.inference_mode():
+        out_ids = lm.generate(
+            **inputs,
+            max_new_tokens=2048,
+            do_sample=True,
+            temperature=0.2,
+            repetition_penalty=1.1,
+            top_k=100,
+            top_p=0.95,
+        )
+
+    return tok.decode(
+        out_ids[0][inputs.input_ids.shape[-1]:], skip_special_tokens=True
+    )
+
+
+INITIALIZATION_STATUS = {"model_loaded": True, "error": None}
+
+
+class GenerateRequest(BaseModel):
+    audio_data: str = Field(..., description="")
+    sample_rate: int = Field(..., description="")
+
+
+class GenerateResponse(BaseModel):
+    audio_data: str = Field(..., description="")
+
+
+app = FastAPI(title="V1", version="0.1")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+def b64(b64: str) -> np.ndarray:
+    raw = base64.b64decode(b64)
+    return np.load(io.BytesIO(raw), allow_pickle=False)
+
+
+def ab64(arr: np.ndarray, sr: int) -> str:
+    buf = io.BytesIO()
+    resampled = librosa.resample(arr, orig_sr=16000, target_sr=sr)
+    np.save(buf, resampled.astype(np.float32))
+    return base64.b64encode(buf.getvalue()).decode()
+
+
+@app.get("/api/v1/health")
+def health_check():
+    return {
+        "status": "healthy",
+        "model_loaded": INITIALIZATION_STATUS["model_loaded"],
+        "error": INITIALIZATION_STATUS["error"],
+    }
+
+
+@app.post("/api/v1/v2v", response_model=GenerateResponse)
+def generate_audio(req: GenerateRequest):
+    print("=== V2V Request Started ===")
+    audio_np = b64(req.audio_data)
+    if audio_np.ndim == 1:
+        audio_np = audio_np.reshape(1, -1)
+    print(f"Audio shape: {audio_np.shape}, Sample rate: {req.sample_rate}")
+
+    system_prompt = (
+        "You are a helpful assistant who tries to help answer the user's question. "
+        "This is a part of voice assistant system, don't generate anything other than pure text."
+    )
+
+    try:
+        text = gt(audio_np, req.sample_rate)
+        response_text = chat(system_prompt, user_prompt=text)
+        print(f"LLM response len chars: '{len(response_text)}'")
+        print(f"LLM response: '{response_text}'")
+
+        import time
+        start_time = time.perf_counter()
+        audio_out = tts.generate(
+            text=response_text,
+            prompt_wav_path=None,
+            prompt_text=None,
+            cfg_value=2.0,
+            inference_timesteps=10,
+            normalize=True,
+            denoise=True,
+            retry_badcase=True,
+            retry_badcase_max_times=3,
+            retry_badcase_ratio_threshold=6.0,
+        )
+        print("TTS generation complete.")
+        end_time = time.perf_counter()
+        print(f"TTS generation took {end_time - start_time:.2f} seconds.")
+        print("=== V2V Request Complete ===")
+    except Exception as e:
+        print(f"ERROR in V2V: {e}")
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=f"{e}")
+
+    return GenerateResponse(audio_data=ab64(audio_out, req.sample_rate))
+
+
+@app.post("/api/v1/v2t")
+def generate_text(req: GenerateRequest):
+    audio_np = b64(req.audio_data)
+    if audio_np.ndim == 1:
+        audio_np = audio_np.reshape(1, -1)
+
+    try:
+        text = gt(audio_np, req.sample_rate)
+        print(f"Transcribed text: {text}")
+        system_prompt = "You are a helpful assistant who tries to help answer the user's question."
+        response_text = chat(system_prompt, user_prompt=text)
+    except Exception as e:
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=f"{e}")
+
+    return {"text": response_text}
+
+
+if __name__ == "__main__":
+    uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=False)

spk_001.wav

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:79de3a5775f8880c0bf3e950b103f03b257db630224fab265a309d82753b1aa5
+size 480044