communex._common

  1import random
  2import re
  3import warnings
  4from collections import defaultdict
  5from enum import Enum
  6from typing import Any, Callable, Mapping, TypeVar
  7
  8from pydantic import SecretStr
  9from pydantic_settings import BaseSettings, SettingsConfigDict
 10
 11from communex.balance import from_nano
 12from communex.types import Ss58Address
 13
 14IPFS_REGEX = re.compile(r"^Qm[1-9A-HJ-NP-Za-km-z]{44}$")
 15
 16
 17def deprecated(func: Callable[..., Any]) -> Callable[..., Any]:
 18    def wrapper(*args: Any, **kwargs: Any) -> Any:
 19        warnings.warn(
 20            f"The function {func.__name__} is deprecated and may be removed in a future version.",
 21            DeprecationWarning,
 22        )
 23        return func(*args, **kwargs)
 24
 25    return wrapper
 26
 27
 28class ComxSettings(BaseSettings):
 29    model_config = SettingsConfigDict(env_prefix="COMX_")
 30    # TODO: improve node lists
 31    NODE_URLS: list[str] = [
 32        "wss://api.communeai.net",
 33    ]
 34    TESTNET_NODE_URLS: list[str] = ["wss://testnet.api.communeai.net"]
 35    UNIVERSAL_PASSWORD: SecretStr | None = None
 36    KEY_PASSWORDS: dict[str, SecretStr] | None = None
 37
 38
 39def get_node_url(
 40    comx_settings: ComxSettings | None = None, *, use_testnet: bool = False
 41) -> str:
 42    comx_settings = comx_settings or ComxSettings()
 43    match use_testnet:
 44        case True:
 45            node_url = random.choice(comx_settings.TESTNET_NODE_URLS)
 46        case False:
 47            node_url = random.choice(comx_settings.NODE_URLS)
 48    return node_url
 49
 50
 51def get_available_nodes(
 52    comx_settings: ComxSettings | None = None, *, use_testnet: bool = False
 53) -> list[str]:
 54    comx_settings = comx_settings or ComxSettings()
 55
 56    match use_testnet:
 57        case True:
 58            node_urls = comx_settings.TESTNET_NODE_URLS
 59        case False:
 60            node_urls = comx_settings.NODE_URLS
 61    return node_urls
 62
 63
 64class BalanceUnit(str, Enum):
 65    joule = "joule"
 66    j = "j"
 67    nano = "nano"
 68    n = "n"
 69
 70
 71def format_balance(balance: int, unit: BalanceUnit = BalanceUnit.nano) -> str:
 72    """
 73    Formats a balance.
 74    """
 75
 76    match unit:
 77        case BalanceUnit.nano | BalanceUnit.n:
 78            return f"{balance}"
 79        case BalanceUnit.joule | BalanceUnit.j:
 80            in_joules = from_nano(balance)
 81            round_joules = round(in_joules, 4)
 82            return f"{round_joules:,} COMAI"
 83
 84
 85K = TypeVar("K")
 86V = TypeVar("V")
 87Z = TypeVar("Z")
 88
 89
 90def intersection_update(
 91    base: dict[K, V], update: dict[K, Z]
 92) -> Mapping[K, V | Z]:
 93    """
 94    Update a dictionary with another dictionary, but only with keys that are already present.
 95    """
 96    updated = {k: update[k] for k in base if k in update}
 97    return updated
 98
 99
100def transform_stake_dmap(
101    stake_storage: dict[tuple[Ss58Address, Ss58Address], int],
102) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]:
103    """
104    Transforms either the StakeTo or StakeFrom storage into the stake legacy data type.
105    """
106    transformed: dict[Ss58Address, list[tuple[Ss58Address, int]]] = defaultdict(
107        list
108    )
109    [transformed[k1].append((k2, v)) for (k1, k2), v in stake_storage.items()]
110
111    return dict(transformed)