Skip to content

Commit

Permalink
[Tests] AsyncSubtensor (Part 4) (#2410)
Browse files Browse the repository at this point in the history
* added tests from `AsyncSubtensor.get_netuids_for_hotkey` until `AsyncSubtensor.neurons_lite`

* ruff

* move out `_decode_hex_identity_dict` from inner function

* added tests until `AsyncSubtensor.query_identity`

* added tests from `AsyncSubtensor.get_netuids_for_hotkey` until `AsyncSubtensor.neurons_lite`

* fix

* ruff
  • Loading branch information
roman-opentensor authored Nov 13, 2024
1 parent 247d92a commit 8aefb55
Show file tree
Hide file tree
Showing 2 changed files with 337 additions and 26 deletions.
53 changes: 27 additions & 26 deletions bittensor/core/async_subtensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,32 @@ def decode_ss58_tuples(line: tuple):
return [decode_account_id(line[x][0]) for x in range(len(line))]


def _decode_hex_identity_dict(info_dictionary: dict[str, Any]) -> dict[str, Any]:
"""Decodes a dictionary of hexadecimal identities."""
for k, v in info_dictionary.items():
if isinstance(v, dict):
item = next(iter(v.values()))
else:
item = v
if isinstance(item, tuple) and item:
if len(item) > 1:
try:
info_dictionary[k] = (
bytes(item).hex(sep=" ", bytes_per_sep=2).upper()
)
except UnicodeDecodeError:
logging.error(f"Could not decode: {k}: {item}.")
else:
try:
info_dictionary[k] = bytes(item[0]).decode("utf-8")
except UnicodeDecodeError:
logging.error(f"Could not decode: {k}: {item}.")
else:
info_dictionary[k] = item

return info_dictionary


class AsyncSubtensor:
"""Thin layer for interacting with Substrate Interface. Mostly a collection of frequently-used calls."""

Expand Down Expand Up @@ -858,7 +884,6 @@ async def neuron_for_uid(
method="neuronInfo_getNeuron",
params=params, # custom rpc method
)

if not (result := json_body.get("result", None)):
return NeuronInfo.get_null_neuron()

Expand Down Expand Up @@ -924,30 +949,6 @@ async def query_identity(
See the `Bittensor CLI documentation <https://docs.bittensor.com/reference/btcli>`_ for supported identity parameters.
"""

def decode_hex_identity_dict_(info_dictionary):
for k, v in info_dictionary.items():
if isinstance(v, dict):
item = next(iter(v.values()))
else:
item = v
if isinstance(item, tuple) and item:
if len(item) > 1:
try:
info_dictionary[k] = (
bytes(item).hex(sep=" ", bytes_per_sep=2).upper()
)
except UnicodeDecodeError:
print(f"Could not decode: {k}: {item}")
else:
try:
info_dictionary[k] = bytes(item[0]).decode("utf-8")
except UnicodeDecodeError:
print(f"Could not decode: {k}: {item}")
else:
info_dictionary[k] = item

return info_dictionary

identity_info = await self.substrate.query(
module="Registry",
storage_function="IdentityOf",
Expand All @@ -956,7 +957,7 @@ def decode_hex_identity_dict_(info_dictionary):
reuse_block_hash=reuse_block,
)
try:
return decode_hex_identity_dict_(identity_info["info"])
return _decode_hex_identity_dict(identity_info["info"])
except TypeError:
return {}

Expand Down
Loading

0 comments on commit 8aefb55

Please sign in to comment.