Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 34 additions & 23 deletions bittensor/core/async_subtensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -3030,19 +3030,22 @@ async def set_reveal_commitment(
# increase reveal_round in return + 1 because we want to fetch data from the chain after that round was revealed
# and stored.
data_ = {"encrypted": encrypted, "reveal_round": reveal_round}
return await publish_metadata(
subtensor=self,
wallet=wallet,
netuid=netuid,
data_type="TimelockEncrypted",
data=data_,
period=period,
), reveal_round
return (
await publish_metadata(
subtensor=self,
wallet=wallet,
netuid=netuid,
data_type="TimelockEncrypted",
data=data_,
period=period,
),
reveal_round,
)

async def subnet(
self,
netuid: int,
block: int = None,
block: Optional[int] = None,
block_hash: Optional[str] = None,
reuse_block: bool = False,
) -> Optional[DynamicInfo]:
Expand All @@ -3052,23 +3055,26 @@ async def subnet(
Args:
netuid (int): The unique identifier of the subnet.
block (Optional[int]): The block number to get the subnets at.
block_hash (str): The hash of the blockchain block number for the query.
block_hash (Optional[str]): The hash of the blockchain block number for the query.
reuse_block (bool): Whether to reuse the last-used blockchain block hash.

Returns:
Optional[DynamicInfo]: A DynamicInfo object, containing detailed information about a subnet.
"""
block_hash = await self.determine_block_hash(block, block_hash, reuse_block)

if not block_hash and reuse_block:
block_hash = self.substrate.last_block_hash

query = await self.substrate.runtime_call(
"SubnetInfoRuntimeApi",
"get_dynamic_info",
params=[netuid],
block_hash=block_hash,
)
subnet = DynamicInfo.from_dict(query.decode())
return subnet

if isinstance(decoded := query.decode(), dict):
return DynamicInfo.from_dict(decoded)

async def subnet_exists(
self,
Expand Down Expand Up @@ -3216,7 +3222,9 @@ async def handler(block_data: dict):
return None

current_block = await self.substrate.get_block()
assert current_block is not None, "Failed to retrieve current block"
current_block_hash = current_block.get("header", {}).get("hash")

if block is not None:
target_block = block
else:
Expand Down Expand Up @@ -3314,16 +3322,16 @@ async def get_timestamp(
Returns:
datetime object for the timestamp of the block
"""
unix = (
await self.query_module(
"Timestamp",
"Now",
block=block,
block_hash=block_hash,
reuse_block=reuse_block,
)
).value
return datetime.fromtimestamp(unix / 1000, tz=timezone.utc)
unix = await self.query_module(
"Timestamp",
"Now",
block=block,
block_hash=block_hash,
reuse_block=reuse_block,
)
assert unix is not None, "Failed to retrieve timestamp"
assert isinstance(unix.value, int), "Timestamp value is not an integer"
return datetime.fromtimestamp(unix.value / 1000, tz=timezone.utc)

async def get_subnet_owner_hotkey(
self, netuid: int, block: Optional[int] = None
Expand All @@ -3341,9 +3349,12 @@ async def get_subnet_owner_hotkey(
Returns:
The hotkey of the subnet owner if available; None otherwise.
"""
return await self.query_subtensor(
hotkey = await self.query_subtensor(
name="SubnetOwnerHotkey", params=[netuid], block=block
)
assert hotkey is not None, "Failed to retrieve subnet owner hotkey"
assert isinstance(hotkey.value, str), "Subnet owner hotkey is not a string"
return hotkey.value

async def get_subnet_validator_permits(
self, netuid: int, block: Optional[int] = None
Expand Down
10 changes: 4 additions & 6 deletions bittensor/core/chain_data/info_base.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,15 @@
from dataclasses import dataclass
from typing import Any, TypeVar
from typing import Any, Self

from bittensor.core.errors import SubstrateRequestException

T = TypeVar("T", bound="InfoBase")


@dataclass
class InfoBase:
"""Base dataclass for info objects."""

@classmethod
def from_dict(cls, decoded: dict) -> T:
def from_dict(cls, decoded: dict) -> Self:
try:
return cls._from_dict(decoded)
except KeyError as e:
Expand All @@ -20,9 +18,9 @@ def from_dict(cls, decoded: dict) -> T:
)

@classmethod
def list_from_dicts(cls, any_list: list[Any]) -> list[T]:
def list_from_dicts(cls, any_list: list[Any]) -> list[Self]:
return [cls.from_dict(any_) for any_ in any_list]

@classmethod
def _from_dict(cls, decoded: dict) -> T:
def _from_dict(cls, decoded: dict) -> Self:
return cls(**decoded)
Loading