Subtensor Registry

bittensor/errors.py

@@ -46,6 +46,11 @@ class UnstakeError(ChainTransactionError):
     pass
 
 
+class IdentityError(ChainTransactionError):
+    r"""Error raised when an identity transaction fails."""
+    pass
+
+
 class NominationError(ChainTransactionError):
     r"""Error raised when a nomination transaction fails."""
     pass

bittensor/subtensor.py

@@ -1319,6 +1319,71 @@ def root_set_weights(
             wait_for_finalization=wait_for_finalization,
             prompt=prompt,
         )
+
+    ########################
+    #### Registry Calls ####
+    ########################
+
+    """ Queries subtensor registry named storage with params and block. """
+
+    def query_identity(
+        self,
+        key: str,
+        block: Optional[int] = None,
+    ) -> Optional[object]:
+        @retry(delay=2, tries=3, backoff=2, max_delay=4)
+        def make_substrate_call_with_retry():
+            with self.substrate as substrate:
+                return substrate.query(
+                    module="Registry",
+                    storage_function='IdentityOf',
+                    params=[key],
+                    block_hash=None
+                    if block == None
+                    else substrate.get_block_hash(block),
+                )
+
+        identity_info  = make_substrate_call_with_retry()
+        return bittensor.utils.wallet_utils.decode_hex_identity_dict(identity_info.value['info'])  
+
+    def update_identity(
+        self,
+        wallet: "bittensor.wallet",
+        params: dict = {},
+        wait_for_inclusion: bool = True,
+        wait_for_finalization: bool = False,
+    ) -> bool:
+        """
+        Creates an identity extrinsics with the specific structure.
+        """
+        call_params = bittensor.utils.wallet_utils.create_identity_dict(**params)
+
+        @retry(delay=2, tries=3, backoff=2, max_delay=4)
+        def make_substrate_call_with_retry():
+            with self.substrate as substrate:
+                call = substrate.compose_call(
+                    call_module="Registry",
+                    call_function="set_identity",
+                    call_params=call_params,
+                )
+                extrinsic = substrate.create_signed_extrinsic(
+                    call=call, keypair=wallet.coldkey
+                )
+                response = substrate.submit_extrinsic(
+                    extrinsic,
+                    wait_for_inclusion=wait_for_inclusion,
+                    wait_for_finalization=wait_for_finalization,
+                )
+                # We only wait here if we expect finalization.
+                if not wait_for_finalization and not wait_for_inclusion:
+                    return True
+                response.process_events()
+                if response.is_success:
+                    return True
+                else:
+                    raise IdentityError(response.error_message)
+
+        return make_substrate_call_with_retry()
 
     ########################
     #### Standard Calls ####

bittensor/utils/wallet_utils.py

@@ -18,7 +18,7 @@
 # DEALINGS IN THE SOFTWARE.
 
 from substrateinterface.utils import ss58
-from typing import Union
+from typing import Union, Optional
 
 from .. import __ss58_format__
 from substrateinterface import Keypair as Keypair
@@ -102,3 +102,61 @@ def is_valid_bittensor_address_or_public_key(address: Union[str, bytes]) -> bool
     else:
         # Invalid address type
         return False
+
+
+
+def create_identity_dict(
+    display: str = "",
+    legal: str = "",
+    web: str = "",
+    riot: str = "",
+    email: str = "",
+    pgp_fingerprint: Optional[str] = None,
+    image: str = "",
+    info: str = "",
+    twitter: str = "",
+) -> dict:
+    """
+    Creates a dictionary with structure for identity extrinsic.
+
+    Args:
+    display (str): String to be converted and stored under 'display'.
+    legal (str): String to be converted and stored under 'legal'.
+    web (str): String to be converted and stored under 'web'.
+    riot (str): String to be converted and stored under 'riot'.
+    email (str): String to be converted and stored under 'email'.
+    pgp_fingerprint (str): String to be converted and stored under 'pgp_fingerprint'. If None, it remains None.
+    image (str): String to be converted and stored under 'image'.
+    info (str): String to be converted and stored under 'info'.
+    twitter (str): String to be converted and stored under 'twitter'.
+
+    Returns:
+    dict: A dictionary with the specified structure and byte string conversions.
+    """
+    return {
+        'info': {
+            'additional': [[]],
+            'display': {f'Raw{len(display.encode())}': display.encode()},
+            'legal': {f'Raw{len(legal.encode())}': legal.encode()},
+            'web': {f'Raw{len(web.encode())}': web.encode()},
+            'riot': {f'Raw{len(riot.encode())}': riot.encode()},
+            'email': {f'Raw{len(email.encode())}': email.encode()},
+            'pgp_fingerprint': pgp_fingerprint.encode() if pgp_fingerprint else None,
+            'image': {f'Raw{len(image.encode())}': image.encode()},
+            'info': {f'Raw{len(info.encode())}': info.encode()},
+            'twitter': {f'Raw{len(twitter.encode())}': twitter.encode()},
+        }
+    }
+
+def decode_hex_identity_dict(info_dictionary):
+    for key, value in info_dictionary.items():
+        if isinstance(value, dict):
+            item = list(value.values())[0]
+            if isinstance(item, str) and item.startswith('0x'):
+                try:
+                    info_dictionary[key] = bytes.fromhex(item[2:]).decode()
+                except UnicodeDecodeError:
+                    print(f"Could not decode: {key}: {item}")
+            else:
+                info_dictionary[key] = item
+    return info_dictionary