Implement web panel to manage hyperparams

app/envs/prod/gunicorn.conf.py

@@ -4,6 +4,3 @@
 bind = "0.0.0.0:8000"
 wsgi_app = "bittensor_panel.wsgi:application"
 access_logfile = "-"
-
-
-

app/src/bittensor_panel/asgi.py

@@ -1,9 +1,6 @@
 import os
 
-
 from django.core.asgi import get_asgi_application
 
-
-
 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "bittensor_panel.settings")
 application = get_asgi_application()

app/src/bittensor_panel/celery.py

@@ -1,8 +1,11 @@
 import logging
 import os
 
-from celery import Celeryfrom django.conf import settings
+from celery import Celery
+from celery.signals import setup_logging
+from django.conf import settings
 from django_structlog.celery.steps import DjangoStructLogInitStep
+
 from .settings import configure_structlog
 
 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "bittensor_panel.settings")

app/src/bittensor_panel/core/admin.py

@@ -1,5 +1,53 @@
-from django.contrib import admin  # noqa
-from django.contrib.admin import register  # noqa
+from django.contrib import admin, messages
+from django.contrib.admin import register
+from django.http.response import (
+    HttpResponseRedirect,
+)
+from django.urls import path, reverse
+
+from .exceptions import BittensorAPIError, HyperParameterUpdateFailed
+from .models import HyperParameter
+from .services import (
+    refresh_hyperparams,
+    update_hyperparam,
+)
+
+
+@register(HyperParameter)
+class HyperParameterAdmin(admin.ModelAdmin):
+    list_display = ["name", "value", "created_at", "updated_at"]
+
+    def has_add_permission(self, request):
+        return False
+
+    def has_delete_permission(self, request, obj=None):
+        return False
+
+    def save_model(self, request, obj, form, change):
+        try:
+            update_hyperparam(obj)
+        except (BittensorAPIError, HyperParameterUpdateFailed) as e:
+            messages.error(request, str(e))
+
+    def get_urls(self):
+        urls = [
+            path(
+                "refresh-hyperparams/",
+                self.admin_site.admin_view(self.refresh_hyperparams_view),
+                name="refresh_hyperparams",
+            ),
+        ]
+        urls += super().get_urls()
+        return urls
+
+    def refresh_hyperparams_view(self, request):
+        if request.method == "POST":
+            try:
+                refresh_hyperparams()
+            except BittensorAPIError as e:
+                messages.error(request, str(e))
+
+        return HttpResponseRedirect(reverse("admin:core_hyperparameter_changelist"))
 
 
 admin.site.site_header = "Bittensor Administration Panel"

app/src/bittensor_panel/core/bt.py

@@ -0,0 +1,55 @@
+from dataclasses import asdict
+
+import bittensor
+from django.conf import settings
+
+from .exceptions import (
+    SubtensorConnectionError,
+    SubtensorServerError,
+)
+
+
+def get_subtensor() -> bittensor.subtensor:
+    try:
+        return bittensor.subtensor(settings.SUBTENSOR_ADDRESS)
+    except (SystemExit, Exception) as e:
+        raise SubtensorConnectionError(f"Failed to connect to subtensor at: {settings.SUBTENSOR_ADDRESS}") from e
+
+
+def get_wallet() -> bittensor.wallet:
+    return bittensor.wallet(
+        name=settings.WALLET_NAME,
+        path=settings.WALLET_PATH,
+    )
+
+
+def load_hyperparams() -> dict[str, int] | None:
+    st = get_subtensor()
+
+    try:
+        hyperparams = st.get_subnet_hyperparameters(settings.SUBNET_UID)
+    except Exception as e:
+        raise SubtensorServerError(f"Failed to load hyperparameters from subtensor\n{e}") from e
+
+    if not hyperparams:
+        return None
+
+    return asdict(hyperparams)
+
+
+def update_remote_hyperparam(name: str, value: str) -> bool:
+    st = get_subtensor()
+    wallet = get_wallet()
+
+    try:
+        return st.set_hyperparameter(
+            wallet=wallet,
+            netuid=settings.SUBNET_UID,
+            parameter=name,
+            value=value,
+            wait_for_inclusion=False,
+            wait_for_finalization=True,
+            prompt=False,
+        )
+    except Exception as exc:
+        raise SubtensorServerError(f"Failed to update hyperparameter in subtensor\n{exc}") from exc

app/src/bittensor_panel/core/exceptions.py

@@ -0,0 +1,14 @@
+class BittensorAPIError(Exception):
+    """Base class for all Bittensor API errors."""
+
+
+class SubtensorConnectionError(BittensorAPIError):
+    """Failed to connect to subtensor instance."""
+
+
+class SubtensorServerError(BittensorAPIError):
+    """Subtensor network returned an error."""
+
+
+class HyperParameterUpdateFailed(Exception):
+    """Failed to update hyperparameter in subtensor."""

app/src/bittensor_panel/core/migrations/0001_initial.py

@@ -0,0 +1,30 @@
+# Generated by Django 4.2.11 on 2024-05-06 18:02
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="HyperParameter",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("name", models.CharField(max_length=255, unique=True)),
+                ("value", models.CharField(max_length=255)),
+            ],
+        ),
+    ]

app/src/bittensor_panel/core/models.py

@@ -1 +1,8 @@
 from django.db import models  # noqa
+
+
+class HyperParameter(models.Model):
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+    name = models.CharField(max_length=255, unique=True)
+    value = models.CharField(max_length=255)

app/src/bittensor_panel/core/services.py

@@ -0,0 +1,34 @@
+from bittensor_panel.core.bt import load_hyperparams, update_remote_hyperparam
+from bittensor_panel.core.exceptions import HyperParameterUpdateFailed
+from bittensor_panel.core.models import HyperParameter
+
+
+def update_hyperparam(instance: HyperParameter) -> None:
+    """
+    Update hyperparameter in the subtensor with the new value and save changes to the database.
+    """
+
+    result = update_remote_hyperparam(instance.name, instance.value)
+
+    if not result:
+        raise HyperParameterUpdateFailed("Failed to update remote hyperparameter. Subtensor returned False.")
+
+    instance.save(update_fields=["value"])
+
+
+def refresh_hyperparams() -> None:
+    """
+    Load hyperparameters from the subtensor and overwrite corresponding records in the database.
+    """
+
+    hyperparam_dict = load_hyperparams()
+
+    if not hyperparam_dict:
+        return
+
+    objs: list[HyperParameter] = []
+
+    for name, value in hyperparam_dict.items():
+        objs.append(HyperParameter(name=name, value=value))
+
+    HyperParameter.objects.bulk_create(objs, update_conflicts=True, update_fields=["value"], unique_fields=["name"])

app/src/bittensor_panel/core/tests/settings.py

@@ -1,6 +1,10 @@
 import os
 
 os.environ["DEBUG_TOOLBAR"] = "False"
+os.environ["SUBTENSOR_ADDRESS"] = "local"
+os.environ["SUBNET_UID"] = "1"
+os.environ["WALLET_NAME"] = "default"
+os.environ["WALLET_HOTKEY"] = "default"
+os.environ["WALLET_PATH"] = ""
 
 from bittensor_panel.settings import *  # noqa: E402,F403
-

app/src/bittensor_panel/core/tests/test_admin.py

@@ -0,0 +1,30 @@
+from unittest.mock import MagicMock
+
+import pytest
+from django.test import Client
+from pytest_mock import MockerFixture
+
+from bittensor_panel.core.exceptions import SubtensorConnectionError
+
+
+@pytest.fixture
+def mock_refresh_hyperparams(mocker: MockerFixture):
+    return mocker.patch("bittensor_panel.core.admin.refresh_hyperparams")
+
+
+def test_refresh_hyperparams_view(admin_client: Client, mock_refresh_hyperparams: MagicMock):
+    response = admin_client.post("/admin/core/hyperparameter/refresh-hyperparams/")
+
+    mock_refresh_hyperparams.assert_called_once()
+
+    assert response.status_code == 302
+    assert response.url == "/admin/core/hyperparameter/"  # type: ignore
+
+
+def test_refresh_hyperparams_view_exception(admin_client: Client, mock_refresh_hyperparams: MagicMock):
+    mock_refresh_hyperparams.side_effect = SubtensorConnectionError
+
+    response = admin_client.post("/admin/core/hyperparameter/refresh-hyperparams/")
+
+    assert response.status_code == 302
+    assert response.url == "/admin/core/hyperparameter/"  # type: ignore