More stripping of py-substrate-interface

bittensor/core/async_subtensor.py

@@ -14,7 +14,6 @@
 from scalecodec import GenericCall, ScaleType
 from scalecodec.base import RuntimeConfiguration
 from scalecodec.type_registry import load_type_registry_preset
-from substrateinterface.exceptions import SubstrateRequestException
 
 from bittensor.core import settings
 from bittensor.core.chain_data import (
@@ -31,6 +30,7 @@
 )
 
 from bittensor.core.config import Config
+from bittensor.core.errors import SubstrateRequestException
 from bittensor.core.extrinsics.asyncex.commit_reveal import commit_reveal_v3_extrinsic
 from bittensor.core.extrinsics.asyncex.registration import (
     burned_register_extrinsic,

bittensor/core/dendrite.py

@@ -61,7 +61,7 @@ class DendriteMixin:
     network requests and processing server responses.
 
     Args:
-        keypair (Option[Union[bittensor_wallet.Wallet, substrateinterface.Keypair]]): The wallet or keypair used for signing messages.
+        keypair (Option[Union[bittensor_wallet.Wallet, bittensor_wallet.Keypair]]): The wallet or keypair used for signing messages.
         external_ip (str): The external IP address of the local system.
         synapse_history (list): A list of Synapse objects representing the historical responses.
 
@@ -104,7 +104,7 @@ def __init__(self, wallet: Optional[Union["Wallet", "Keypair"]] = None):
         Initializes the Dendrite object, setting up essential properties.
 
         Args:
-            wallet (Optional[Union[bittensor_wallet.Wallet, substrateinterface.Keypair]]): The user's wallet or keypair used for signing messages. Defaults to ``None``, in which case a new :func:`bittensor_wallet.Wallet().hotkey` is generated and used.
+            wallet (Optional[Union[bittensor_wallet.Wallet, bittensor_wallet..Keypair]]): The user's wallet or keypair used for signing messages. Defaults to ``None``, in which case a new :func:`bittensor_wallet.Wallet().hotkey` is generated and used.
         """
         # Initialize the parent class
         super(DendriteMixin, self).__init__()

bittensor/core/extrinsics/asyncex/root.py

@@ -5,8 +5,8 @@
 import numpy as np
 from bittensor_wallet import Wallet
 from numpy.typing import NDArray
-from substrateinterface.exceptions import SubstrateRequestException
 
+from bittensor.core.errors import SubstrateRequestException
 from bittensor.utils import u16_normalized_float, format_error_message, unlock_key
 from bittensor.utils.btlogging import logging
 from bittensor.utils.weight_utils import (

bittensor/core/extrinsics/utils.py

@@ -9,8 +9,10 @@
 if TYPE_CHECKING:
     from bittensor.core.subtensor import Subtensor
     from bittensor.core.async_subtensor import AsyncSubtensor
-    from bittensor.utils.substrate_interface import AsyncExtrinsicReceipt
-    from substrateinterface import ExtrinsicReceipt
+    from bittensor.utils.substrate_interface import (
+        AsyncExtrinsicReceipt,
+        ExtrinsicReceipt,
+    )
     from scalecodec.types import GenericExtrinsic
 
 

bittensor/utils/__init__.py

@@ -17,14 +17,15 @@
 
 import asyncio
 import ast
+import base58
 from collections import namedtuple
 import hashlib
+from hashlib import blake2b
 from typing import Any, Literal, Union, Optional, TYPE_CHECKING, Coroutine
 from urllib.parse import urlparse
 
 import scalecodec
 from bittensor_wallet import Keypair
-from substrateinterface.utils import ss58
 
 from bittensor.core.settings import SS58_FORMAT
 from bittensor.utils.btlogging import logging
@@ -53,6 +54,108 @@
 UnlockStatus = namedtuple("UnlockStatus", ["success", "message"])
 
 
+def ss58_decode(address: str, valid_ss58_format: Optional[int] = None) -> str:
+    """
+    Decodes given SS58 encoded address to an account ID
+
+    Args:
+        address: e.g. EaG2CRhJWPb7qmdcJvy3LiWdh26Jreu9Dx6R1rXxPmYXoDk
+        valid_ss58_format: the format for what is considered valid
+
+    Returns:
+        Decoded string AccountId
+    """
+
+    # Check if address is already decoded
+    if address.startswith("0x"):
+        return address
+
+    if address == "":
+        raise ValueError("Empty address provided")
+
+    checksum_prefix = b"SS58PRE"
+
+    address_decoded = base58.b58decode(address)
+
+    if address_decoded[0] & 0b0100_0000:
+        ss58_format_length = 2
+        ss58_format = (
+            ((address_decoded[0] & 0b0011_1111) << 2)
+            | (address_decoded[1] >> 6)
+            | ((address_decoded[1] & 0b0011_1111) << 8)
+        )
+    else:
+        ss58_format_length = 1
+        ss58_format = address_decoded[0]
+
+    if ss58_format in [46, 47]:
+        raise ValueError(f"{ss58_format} is a reserved SS58 format")
+
+    if valid_ss58_format is not None and ss58_format != valid_ss58_format:
+        raise ValueError("Invalid SS58 format")
+
+    # Determine checksum length according to length of address string
+    if len(address_decoded) in [3, 4, 6, 10]:
+        checksum_length = 1
+    elif len(address_decoded) in [
+        5,
+        7,
+        11,
+        34 + ss58_format_length,
+        35 + ss58_format_length,
+    ]:
+        checksum_length = 2
+    elif len(address_decoded) in [8, 12]:
+        checksum_length = 3
+    elif len(address_decoded) in [9, 13]:
+        checksum_length = 4
+    elif len(address_decoded) in [14]:
+        checksum_length = 5
+    elif len(address_decoded) in [15]:
+        checksum_length = 6
+    elif len(address_decoded) in [16]:
+        checksum_length = 7
+    elif len(address_decoded) in [17]:
+        checksum_length = 8
+    else:
+        raise ValueError("Invalid address length")
+
+    checksum = blake2b(checksum_prefix + address_decoded[0:-checksum_length]).digest()
+
+    if checksum[0:checksum_length] != address_decoded[-checksum_length:]:
+        raise ValueError("Invalid checksum")
+
+    return address_decoded[
+        ss58_format_length : len(address_decoded) - checksum_length
+    ].hex()
+
+
+def _is_valid_ss58_address(value: str, valid_ss58_format: Optional[int] = None) -> bool:
+    """
+    Checks if given value is a valid SS58 formatted address, optionally check if address is valid for specified
+    ss58_format
+
+    Args:
+        value: value to checked
+        valid_ss58_format: if valid_ss58_format is provided the address must be valid for specified ss58_format
+            (network) as well
+
+    Returns:
+        bool result
+    """
+
+    # Return False in case a public key is provided
+    if value.startswith("0x"):
+        return False
+
+    try:
+        ss58_decode(value, valid_ss58_format=valid_ss58_format)
+    except ValueError:
+        return False
+
+    return True
+
+
 def event_loop_is_running() -> Optional[asyncio.AbstractEventLoop]:
     """
     Simple function to check if event loop is running. Returns the loop if it is, otherwise None.
@@ -247,9 +350,9 @@ def is_valid_ss58_address(address: str) -> bool:
         True if the address is a valid ss58 address for Bittensor, False otherwise.
     """
     try:
-        return ss58.is_valid_ss58_address(
+        return _is_valid_ss58_address(
             address, valid_ss58_format=SS58_FORMAT
-        ) or ss58.is_valid_ss58_address(
+        ) or _is_valid_ss58_address(
             address, valid_ss58_format=42
         )  # Default substrate ss58 format (legacy)
     except IndexError:
@@ -315,7 +418,8 @@ def decode_hex_identity_dict(info_dictionary) -> dict[str, Any]:
     """
     Decodes hex-encoded strings in a dictionary.
 
-    This function traverses the given dictionary, identifies hex-encoded strings, and decodes them into readable strings. It handles nested dictionaries and lists within the dictionary.
+    This function traverses the given dictionary, identifies hex-encoded strings, and decodes them into readable
+        strings. It handles nested dictionaries and lists within the dictionary.
 
     Args:
         info_dictionary (dict): The dictionary containing hex-encoded strings to decode.
@@ -439,7 +543,7 @@ def execute_coroutine(
     Args:
         coroutine (Coroutine): The coroutine to run.
         event_loop (AbstractEventLoop): The event loop to use. If `None`, attempts to fetch the already-running
-            loop. If one if not running, a new loop is created.
+            loop. If one is not running, a new loop is created.
 
     Returns:
         The result of the coroutine execution.

bittensor/utils/registration/async_pow.py

@@ -7,7 +7,7 @@
 from typing import Callable, Union, Optional, TYPE_CHECKING
 
 from retry import retry
-from substrateinterface.exceptions import SubstrateRequestException
+from bittensor.core.errors import SubstrateRequestException
 
 from bittensor.utils.registration.pow import (
     get_cpu_count,

requirements/prod.txt

@@ -3,6 +3,7 @@ setuptools~=70.0.0
 aiohttp~=3.9
 asyncstdlib~=3.13.0
 bittensor-cli
+base58
 bt-decode==0.4.0
 colorama~=0.4.6
 fastapi~=0.110.1
@@ -21,6 +22,7 @@ rich
 pydantic>=2.3, <3
 python-Levenshtein
 scalecodec==1.2.11
+substrate-interface~=1.7.9  # still needed for StorageKey in substrate_interface.py
 uvicorn
 websockets>=14.1
 bittensor-wallet>=2.1.3