Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • hive/clive
1 result
Show changes
Commits on Source (18)
Showing
with 643 additions and 29 deletions
......@@ -123,7 +123,10 @@ testing_clive:
extends: .testing
script:
- echo -e "${TXT_BLUE}Launching clive concurrent tests...${TXT_CLEAR}"
- python3 -m pytest --ignore "tests/functional/beekeeper" --ignore "tests/tui" -k "not test_autocompletion_time" -n auto --durations 0 --junitxml=report.xml tests/
- python3 -m pytest
--ignore "tests/functional/beekeeper" --ignore "tests/tui" --ignore "tests/functional/cli"
-k "not test_autocompletion_time"
-n auto --durations 0 --junitxml=report.xml tests/
testing_clive_import_times_during_autocompletion:
extends: .testing
......@@ -137,6 +140,12 @@ testing_tui:
- echo -e "${TXT_BLUE}Launching tui tests...${TXT_CLEAR}"
- python3 -m pytest -n auto --durations 0 --junitxml=report.xml tests/tui
testing_cli:
extends: .testing
script:
- echo -e "${TXT_BLUE}Launching cli commands tests...${TXT_CLEAR}"
- python3 -m pytest -n auto --durations 0 --junitxml=report.xml tests/functional/cli
testing_password_private_key_logging:
stage: tests
needs:
......
from __future__ import annotations
from math import ceil
from typing import TYPE_CHECKING
from clive.__private.core.vests_to_hive import vests_to_hive
if TYPE_CHECKING:
from clive.models import Asset
from clive.models.aliased import DynamicGlobalProperties
def calculate_hive_power(gdpo: DynamicGlobalProperties, vests_value: Asset.Vests | int) -> int:
vests_to_hive_amount = int(vests_to_hive(vests_value, gdpo).amount)
precision = gdpo.total_reward_fund_hive.get_asset_information().precision
return int(ceil(vests_to_hive_amount / (10**precision)))
......@@ -8,6 +8,7 @@ from clive.__private.core.commands.broadcast import Broadcast
from clive.__private.core.commands.build_transaction import BuildTransaction
from clive.__private.core.commands.command_wrappers import CommandWithResultWrapper, CommandWrapper
from clive.__private.core.commands.create_wallet import CreateWallet
from clive.__private.core.commands.data_retrieval.hive_power_data import HivePowerData, HivePowerDataRetrieval
from clive.__private.core.commands.data_retrieval.proposals_data import ProposalsData, ProposalsDataRetrieval
from clive.__private.core.commands.data_retrieval.savings_data import SavingsData, SavingsDataRetrieval
from clive.__private.core.commands.data_retrieval.update_node_data import UpdateNodeData
......@@ -300,6 +301,15 @@ class Commands(Generic[WorldT]):
)
)
async def retrieve_hp_data(
self,
*,
account_name: str,
) -> CommandWithResultWrapper[HivePowerData]:
return await self.__surround_with_exception_handlers(
HivePowerDataRetrieval(node=self._world.node, account_name=account_name)
)
async def find_transaction(self, *, transaction_id: str) -> CommandWithResultWrapper[TransactionStatus]:
return await self.__surround_with_exception_handlers(
FindTransaction(node=self._world.node, transaction_id=transaction_id)
......
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING, Final
from clive.__private.core.commands.abc.command_data_retrieval import CommandDataRetrieval
from clive.__private.core.vests_to_hive import vests_to_hive
from clive.models import Asset
if TYPE_CHECKING:
from datetime import datetime
from clive.__private.core.node import Node
from clive.models.aliased import DynamicGlobalProperties, SchemasAccount
from schemas.apis.database_api import FindAccounts, FindVestingDelegations, ListWithdrawVestingRoutes
from schemas.apis.database_api.fundaments_of_reponses import VestingDelegationsFundament as VestingDelegation
from schemas.apis.database_api.fundaments_of_reponses import WithdrawVestingRoutesFundament as WithdrawRoute
_MAX_WITHDRAW_VESTING_ROUTES_LIMIT: Final[int] = 10
@dataclass
class HarvestedDataRaw:
gdpo: DynamicGlobalProperties | None = None
core_account: FindAccounts | None = None
withdraw_routes: ListWithdrawVestingRoutes | None = None
delegations: FindVestingDelegations | None = None
@dataclass
class SanitizedData:
gdpo: DynamicGlobalProperties
core_account: SchemasAccount
withdraw_routes: list[WithdrawRoute]
delegations: list[VestingDelegation[Asset.Vests]]
@dataclass
class SharesBalance:
"""Class to store the balance of shares in HP and VESTS."""
hp_balance: Asset.Hive
vests_balance: Asset.Vests
@dataclass
class HivePowerData:
owned_balance: SharesBalance
total_balance: SharesBalance
received_balance: SharesBalance
delegated_balance: SharesBalance
next_vesting_withdrawal: datetime
withdraw_routes: list[WithdrawRoute]
delegations: list[VestingDelegation[Asset.Vests]]
to_withdraw: SharesBalance
next_power_down: SharesBalance
current_hp_apr: str
gdpo: DynamicGlobalProperties
@dataclass(kw_only=True)
class HivePowerDataRetrieval(CommandDataRetrieval[HarvestedDataRaw, SanitizedData, HivePowerData]):
node: Node
account_name: str
async def _harvest_data_from_api(self) -> HarvestedDataRaw:
async with self.node.batch() as node:
return HarvestedDataRaw(
await node.api.database_api.get_dynamic_global_properties(),
await node.api.database_api.find_accounts(accounts=[self.account_name]),
await node.api.database_api.list_withdraw_vesting_routes(
start=(self.account_name, ""), limit=_MAX_WITHDRAW_VESTING_ROUTES_LIMIT, order="by_withdraw_route"
),
await node.api.database_api.find_vesting_delegations(account=self.account_name),
)
async def _sanitize_data(self, data: HarvestedDataRaw) -> SanitizedData:
return SanitizedData(
gdpo=self._assert_gdpo(data.gdpo),
core_account=self._assert_core_account(data.core_account),
withdraw_routes=self._assert_withdraw_routes(data.withdraw_routes),
delegations=self._assert_delegations(data.delegations),
)
async def _process_data(self, data: SanitizedData) -> HivePowerData:
owned_shares = data.core_account.vesting_shares
received_shares = data.core_account.received_vesting_shares
delegated_shares = data.core_account.delegated_vesting_shares
total_shares = owned_shares + received_shares - delegated_shares - data.core_account.vesting_withdraw_rate
return HivePowerData(
owned_balance=self._create_balance_representation(data.gdpo, owned_shares),
total_balance=self._create_balance_representation(data.gdpo, total_shares),
received_balance=self._create_balance_representation(data.gdpo, received_shares),
delegated_balance=self._create_balance_representation(data.gdpo, delegated_shares),
next_vesting_withdrawal=data.core_account.next_vesting_withdrawal,
withdraw_routes=[route for route in data.withdraw_routes if route.from_account == self.account_name],
delegations=data.delegations,
to_withdraw=self._create_balance_representation(
data.gdpo, Asset.vests(data.core_account.to_withdraw / 10**data.gdpo.total_vesting_shares.precision)
),
next_power_down=self._create_balance_representation(data.gdpo, data.core_account.vesting_withdraw_rate),
current_hp_apr=self._calculate_current_hp_apr(data.gdpo),
gdpo=data.gdpo,
)
def _assert_gdpo(self, data: DynamicGlobalProperties | None) -> DynamicGlobalProperties:
assert data is not None, "DynamicGlobalProperties data is missing"
return data
def _assert_core_account(self, data: FindAccounts | None) -> SchemasAccount:
assert data is not None, "FindAccounts data is missing"
assert len(data.accounts) == 1, "Invalid amount of accounts"
account = data.accounts[0]
assert account.name == self.account_name, "Invalid account name"
return account
def _assert_withdraw_routes(self, data: ListWithdrawVestingRoutes | None) -> list[WithdrawRoute]:
assert data is not None, "ListWithdrawVestingRoutes data is missing"
return data.routes
def _assert_delegations(self, data: FindVestingDelegations | None) -> list[VestingDelegation[Asset.Vests]]:
assert data is not None, "FindVestingDelegations data is missing"
return data.delegations
def _create_balance_representation(self, gdpo: DynamicGlobalProperties, vests_value: Asset.Vests) -> SharesBalance:
return SharesBalance(hp_balance=vests_to_hive(vests_value, gdpo), vests_balance=vests_value)
def _calculate_current_hp_apr(self, gdpo: DynamicGlobalProperties) -> str:
# The inflation was set to 9.5% at block 7m
initial_inflation_rate: Final[float] = 9.5
initial_block: Final[int] = 7000000
# It decreases by 0.01% every 250k blocks
decrease_rate: Final[int] = 250000
decrease_percent_per_increment: Final[float] = 0.01
# How many increments have happened since block 7m?
head_block = gdpo.head_block_number
delta_blocks = head_block - initial_block
decrease_increments = delta_blocks / decrease_rate
current_inflation_rate = initial_inflation_rate - decrease_increments * decrease_percent_per_increment
# Cannot go lower than 0.95 %
minimum_inflation_rate: Final[float] = 0.95
if current_inflation_rate < minimum_inflation_rate:
current_inflation_rate = 0.95
# Calculate the APR
vesting_reward_percent = gdpo.vesting_reward_percent / 10000
virtual_supply = int(gdpo.virtual_supply.amount) / 10**gdpo.virtual_supply.precision
total_vesting_funds = int(gdpo.total_vesting_fund_hive.amount) / 10**gdpo.total_vesting_fund_hive.precision
return f"{virtual_supply * current_inflation_rate * vesting_reward_percent / total_vesting_funds :.2f}"
......@@ -4,17 +4,17 @@ import re
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from math import ceil
from typing import TYPE_CHECKING, Final, cast
from typing import TYPE_CHECKING, Final
from clive.__private.core.calcluate_hive_power import calculate_hive_power
from clive.__private.core.commands.abc.command_data_retrieval import CommandDataRetrieval
from clive.__private.core.iwax import (
calculate_current_manabar_value,
calculate_manabar_full_regeneration_time,
)
from clive.__private.core.vests_to_hive import vests_to_hive
from clive.__private.storage.accounts import Account
from clive.exceptions import CommunicationError
from clive.models import Asset
from clive.models.aliased import (
DynamicGlobalProperties,
)
......@@ -239,7 +239,7 @@ class UpdateNodeData(CommandDataRetrieval[HarvestedDataRaw, SanitizedData, Dynam
account.data.recovery_account = info.core.recovery_account
account.data.proxy = info.core.proxy
account.data.hp_balance = self.__calculate_hive_power(gdpo, info.core)
account.data.hp_balance = calculate_hive_power(gdpo, self._calculate_vests_balance(info.core))
self.__update_manabar(
gdpo, int(info.core.post_voting_power.amount), info.core.voting_manabar, account.data.vote_manabar
......@@ -302,32 +302,21 @@ class UpdateNodeData(CommandDataRetrieval[HarvestedDataRaw, SanitizedData, Dynam
return _get_utc_epoch()
return self.__normalize_datetime(data.history[0][1].timestamp)
def __calculate_hive_power(
self,
gdpo: DynamicGlobalProperties,
account: SchemasAccount,
) -> int:
account_vesting_shares = (
def _calculate_vests_balance(self, account: SchemasAccount) -> int:
return (
int(account.vesting_shares.amount)
- int(account.delegated_vesting_shares.amount)
+ int(account.received_vesting_shares.amount)
)
return cast(
int,
ceil(
int(self.__vests_to_hive(account_vesting_shares, gdpo).amount)
/ (10 ** gdpo.total_reward_fund_hive.get_asset_information().precision)
),
)
def __update_manabar(
self, gdpo: DynamicGlobalProperties, max_mana: int, manabar: Manabar, dest: mock_database.Manabar
) -> None:
power_from_api = int(manabar.current_mana)
last_update = int(manabar.last_update_time)
dest.max_value = int(self.__vests_to_hive(max_mana, gdpo).amount)
dest.max_value = int(vests_to_hive(max_mana, gdpo).amount)
dest.value = int(
self.__vests_to_hive(
vests_to_hive(
calculate_current_manabar_value(
now=int(gdpo.time.timestamp()),
max_mana=max_mana,
......@@ -357,13 +346,6 @@ class UpdateNodeData(CommandDataRetrieval[HarvestedDataRaw, SanitizedData, Dynam
- gdpo_time
)
def __vests_to_hive(self, amount: int | Asset.Vests, gdpo: DynamicGlobalProperties) -> Asset.Hive:
if isinstance(amount, Asset.Vests):
amount = int(amount.amount)
return Asset.Hive(
amount=int(amount * int(gdpo.total_vesting_fund_hive.amount) / int(gdpo.total_vesting_shares.amount))
)
@staticmethod
def __normalize_datetime(date: datetime) -> datetime:
return date.replace(microsecond=0, tzinfo=timezone.utc)
......
......@@ -2,7 +2,7 @@ from __future__ import annotations
import re
from datetime import datetime
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, Literal
import humanize
import inflection
......@@ -21,7 +21,7 @@ if TYPE_CHECKING:
def _is_null_date(value: datetime) -> bool:
return value == datetime(1970, 1, 1, 0, 0, 0)
return value == datetime(1970, 1, 1, 0, 0, 0) or value == datetime(1969, 12, 31, 23, 59, 59)
def humanize_natural_time(value: datetime | timedelta) -> str:
......@@ -149,3 +149,12 @@ def humanize_votes_with_comma(
"""Return pretty formatted votes converted to hive power."""
hive_power = calculate_hp_from_votes(votes, total_vesting_fund_hive, total_vesting_shares)
return f"{humanize.intcomma(hive_power, ndigits=0)} HP"
def humanize_asset(asset: Asset.AnyT, *, show_symbol: bool = True, sign_prefix: Literal["", "+", "-"] = "") -> str:
pretty_asset = Asset.pretty_amount(asset)
asset_symbol = asset.get_asset_information().symbol[0]
if sign_prefix and int(asset.amount) != 0:
# To not allow display + or - if balance is equal to zero.
return f"{sign_prefix}{pretty_asset} {asset_symbol if show_symbol else ''}".rstrip()
return f"{pretty_asset} {asset_symbol if show_symbol else ''}".rstrip()
from __future__ import annotations
from typing import TYPE_CHECKING
from clive.models import Asset
if TYPE_CHECKING:
from clive.models.aliased import DynamicGlobalProperties
def vests_to_hive(amount: int | Asset.Vests, gdpo: DynamicGlobalProperties) -> Asset.Hive:
if isinstance(amount, Asset.Vests):
amount = int(amount.amount)
return Asset.Hive(
amount=int(amount * int(gdpo.total_vesting_fund_hive.amount) / int(gdpo.total_vesting_shares.amount))
)
......@@ -9,6 +9,7 @@ from textual.worker import Worker, WorkerState
from clive.__private.abstract_class import AbstractClassMessagePump
from clive.__private.config import settings
from clive.__private.ui.widgets.clive_screen import CliveScreen
from clive.__private.ui.widgets.clive_widget import CliveWidget
from clive.exceptions import CliveError
......@@ -90,9 +91,11 @@ class DataProvider(CliveWidget, Generic[ProviderContentT], AbstractClassMessageP
def stop(self) -> None:
self.interval.stop()
@on(CliveScreen.Suspended)
def pause(self) -> None:
self.interval.pause()
@on(CliveScreen.Resumed)
def resume(self) -> None:
self.interval.resume()
......
from __future__ import annotations
from textual import work
from textual.reactive import var
from clive.__private.core.commands.data_retrieval.hive_power_data import HivePowerData
from clive.__private.ui.data_providers.abc.data_provider import DataProvider
class HivePowerDataProvider(DataProvider[HivePowerData]):
_content: HivePowerData | None = var(None, init=False) # type: ignore[assignment]
def __init__(self, *, paused: bool = False, init_update: bool = True) -> None:
super().__init__(paused=paused, init_update=init_update)
@work(name="hive power data update worker")
async def update(self) -> None:
account_name = self.app.world.profile_data.working_account.name
wrapper = await self.app.world.commands.retrieve_hp_data(account_name=account_name)
if wrapper.error_occurred:
self.notify(f"Failed to retrieve hive power data: {wrapper.error}", severity="error")
return
result = wrapper.result_or_raise
self._content = result
from __future__ import annotations
from clive.__private.ui.operations.governance_operations.governance_operations import Governance
from clive.__private.ui.operations.hive_power_management.hive_power_management import HivePowerManagement
from clive.__private.ui.operations.savings_operations.savings_operations import Savings
from clive.__private.ui.operations.transfer_to_account.transfer_to_account import TransferToAccount
......@@ -8,4 +9,5 @@ __all__ = [
"TransferToAccount",
"Savings",
"Governance",
"HivePowerManagement",
]
from __future__ import annotations
from typing import TYPE_CHECKING
from textual.containers import Vertical
from textual.widgets import Static
from clive.__private.core.formatters.humanize import humanize_datetime
from clive.__private.ui.widgets.clive_widget import CliveWidget
from clive.__private.ui.widgets.dynamic_label import DynamicLabel
from clive.models import Asset
if TYPE_CHECKING:
from textual.app import ComposeResult
from clive.__private.core.commands.data_retrieval.hive_power_data import HivePowerData
from clive.__private.ui.data_providers.hive_power_data_provider import HivePowerDataProvider
class WithdrawalInfo(Vertical, CliveWidget):
"""Widget that displays all withdrawal information."""
DEFAULT_CSS = """
WithdrawalInfo {
height: auto;
}
Static {
width: 1fr;
text-align: center;
text-style: bold;
}
.withdrawal-info-header {
background: $primary-background;
}
#to-withdraw-header {
margin-top: 1;
}
#withdrawal-info-date, #withdrawal-info-vests-amount {
background: $accent;
}
#withdrawal-info-hp-amount {
background: $accent-lighten-1;
}
"""
def __init__(self, provider: HivePowerDataProvider):
super().__init__()
self._provider = provider
def compose(self) -> ComposeResult:
yield Static("Next withdrawal", classes="withdrawal-info-header")
yield DynamicLabel(
self._provider, "_content", self._get_next_withdrawal_date, id_="withdrawal-info-date", init=False
)
yield Static("To withdraw", classes="withdrawal-info-header", id="to-withdraw-header")
yield DynamicLabel(
self._provider, "_content", self._get_to_withdraw_hp, id_="withdrawal-info-vests-amount", init=False
)
yield DynamicLabel(
self._provider, "_content", self._get_to_withdraw_vests, id_="withdrawal-info-hp-amount", init=False
)
def _get_next_withdrawal_date(self, content: HivePowerData) -> str:
return humanize_datetime(content.next_vesting_withdrawal)
def _get_to_withdraw_hp(self, content: HivePowerData) -> str:
return f"{Asset.pretty_amount(content.to_withdraw.hp_balance)} HP"
def _get_to_withdraw_vests(self, content: HivePowerData) -> str:
return f"{Asset.pretty_amount(content.to_withdraw.vests_balance)} VESTS"
class APR(DynamicLabel):
DEFAULT_CSS = """
APR {
height: 1;
margin-top: 1;
background: $primary-background;
text-style: bold;
align: center middle;
width: 1fr;
}
"""
def __init__(self, provider: HivePowerDataProvider):
super().__init__(obj_to_watch=provider, attribute_name="_content", callback=self._get_apr, init=False)
self._provider = provider
def _get_apr(self, content: HivePowerData) -> str:
return f"APR interest for HP/VESTS ≈ {content.current_hp_apr} %"
from __future__ import annotations
from typing import TYPE_CHECKING
from textual.containers import Horizontal
from textual.widgets import Static
from clive.__private.core.formatters.humanize import humanize_asset
from clive.__private.ui.data_providers.hive_power_data_provider import HivePowerDataProvider
from clive.__private.ui.widgets.clive_data_table import CliveDataTableRow
if TYPE_CHECKING:
from typing import Final
from textual.app import ComposeResult
from clive.__private.core.commands.data_retrieval.hive_power_data import HivePowerData
class HpInfoTableRow(CliveDataTableRow):
BALANCE_CELL_CLASS: Final[str] = "shares-balance-cell"
def __init__(self, title: str, classes: str | None):
super().__init__(
title,
Static("loading...", classes=self.BALANCE_CELL_CLASS),
Static("loading...", classes=self.BALANCE_CELL_CLASS),
dynamic=True,
classes=classes,
)
@property
def provider(self) -> HivePowerDataProvider:
return self.app.query_one(HivePowerDataProvider)
class HpInfoTableHeader(Horizontal):
def compose(self) -> ComposeResult:
yield Static("Voting Power", id="shares-name-header")
yield Static("Amount in HP", classes="shares-balance-header")
yield Static("Amount in VESTS", classes="shares-balance-header")
class HpInfoTableOwnedRow(HpInfoTableRow):
def __init__(self) -> None:
super().__init__(
"Owned",
classes="odd-row",
)
def get_new_values(self, content: HivePowerData) -> tuple[str, ...]:
hp_balance = humanize_asset(content.owned_balance.hp_balance, show_symbol=False)
vests_balance = humanize_asset(content.owned_balance.vests_balance, show_symbol=False)
return hp_balance, vests_balance
class HpInfoTableReceivedRow(HpInfoTableRow):
def __init__(self) -> None:
super().__init__(
"Received",
classes="even-row",
)
def get_new_values(self, content: HivePowerData) -> tuple[str, ...]:
hp_balance = humanize_asset(content.received_balance.hp_balance, show_symbol=False, sign_prefix="+")
vests_balance = humanize_asset(content.received_balance.vests_balance, show_symbol=False, sign_prefix="+")
return hp_balance, vests_balance
class HpInfoTableDelegatedRow(HpInfoTableRow):
def __init__(self) -> None:
super().__init__(
"Delegated",
classes="odd-row",
)
def get_new_values(self, content: HivePowerData) -> tuple[str, ...]:
hp_balance = humanize_asset(content.delegated_balance.hp_balance, show_symbol=False, sign_prefix="-")
vests_balance = humanize_asset(content.delegated_balance.vests_balance, show_symbol=False, sign_prefix="-")
return hp_balance, vests_balance
class HpInfoTablePowerDownRow(HpInfoTableRow):
def __init__(self) -> None:
super().__init__(
"Power Down",
classes="even-row",
)
def get_new_values(self, content: HivePowerData) -> tuple[str, ...]:
hp_balance = humanize_asset(content.next_power_down.hp_balance, show_symbol=False, sign_prefix="-")
vests_balance = humanize_asset(content.next_power_down.vests_balance, show_symbol=False, sign_prefix="-")
return hp_balance, vests_balance
class HpInfoTableEffectiveRow(HpInfoTableRow):
def __init__(self) -> None:
super().__init__(
"Effective",
classes="odd-row",
)
def get_new_values(self, content: HivePowerData) -> tuple[str, ...]:
hp_balance = humanize_asset(content.total_balance.hp_balance, show_symbol=False)
vests_balance = humanize_asset(content.total_balance.vests_balance, show_symbol=False)
return hp_balance, vests_balance
from __future__ import annotations
from typing import TYPE_CHECKING
from textual.widgets import TabPane
from clive.__private.ui.widgets.clive_widget import CliveWidget
if TYPE_CHECKING:
from rich.text import TextType
class DelegateHivePower(TabPane, CliveWidget):
"""TabPane with all content about delegate hp."""
def __init__(self, title: TextType):
"""
Initialize a TabPane.
Args:
----
title: Title of the TabPane (will be displayed in a tab label).
"""
super().__init__(title=title)
from __future__ import annotations
from typing import TYPE_CHECKING, Final
from textual.containers import Horizontal
from clive.__private.ui.data_providers.hive_power_data_provider import HivePowerDataProvider
from clive.__private.ui.get_css import get_relative_css_path
from clive.__private.ui.operations.hive_power_management.common_hive_power.additional_info_widgets import (
APR,
WithdrawalInfo,
)
from clive.__private.ui.operations.hive_power_management.common_hive_power.hp_information_table import (
HpInfoTableDelegatedRow,
HpInfoTableEffectiveRow,
HpInfoTableHeader,
HpInfoTableOwnedRow,
HpInfoTablePowerDownRow,
HpInfoTableReceivedRow,
)
from clive.__private.ui.operations.hive_power_management.delegate_hive_power.delegate_hive_power import (
DelegateHivePower,
)
from clive.__private.ui.operations.hive_power_management.power_down.power_down import PowerDown
from clive.__private.ui.operations.hive_power_management.power_up.power_up import PowerUp
from clive.__private.ui.operations.operation_base_screen import OperationBaseScreen
from clive.__private.ui.widgets.big_title import BigTitle
from clive.__private.ui.widgets.clive_data_table import CliveDataTable
from clive.__private.ui.widgets.clive_tabbed_content import CliveTabbedContent
if TYPE_CHECKING:
from textual.app import ComposeResult
POWER_UP_TAB_LABEL: Final[str] = "Power up"
POWER_DOWN_TAB_LABEL: Final[str] = "Power down"
DELEGATE_HIVE_POWER_LABEL: Final[str] = "Delegate"
class HivePowerManagement(OperationBaseScreen):
CSS_PATH = [
*OperationBaseScreen.CSS_PATH,
get_relative_css_path(__file__),
]
def create_left_panel(self) -> ComposeResult:
with HivePowerDataProvider() as provider:
yield BigTitle("hive power management")
with Horizontal(id="hive-power-info"):
yield CliveDataTable(
HpInfoTableHeader(),
HpInfoTableOwnedRow(),
HpInfoTableReceivedRow(),
HpInfoTableDelegatedRow(),
HpInfoTablePowerDownRow(),
HpInfoTableEffectiveRow(),
)
yield WithdrawalInfo(provider)
yield APR(provider)
with CliveTabbedContent():
yield PowerUp(POWER_UP_TAB_LABEL)
yield PowerDown(POWER_DOWN_TAB_LABEL)
yield DelegateHivePower(DELEGATE_HIVE_POWER_LABEL)
Tab {
width: 1fr;
}
#hive-power-info {
height: 6;
}
WithdrawalInfo {
width: 1fr;
}
/* Table with amounts */
$row-color-1: $accent;
$row-color-2: $accent-lighten-1;
$table-header-color: $primary-background;
$table-row-title-color: $background-lighten-2;
CliveDataTable {
margin-right: 2;
width: 2fr;
height: auto;
}
/* Header of the table */
HpInfoTableHeader {
height: auto;
}
HpInfoTableHeader Static {
text-style: bold;
text-align: center;
}
HpInfoTableHeader #shares-name-header {
background: $table-header-color;
width: 2fr;
}
HpInfoTableHeader .shares-balance-header {
background: $table-header-color;
width: 3fr;
}
/* Cells of the table */
RowTitle {
text-style: bold;
background: $table-row-title-color;
width: 2fr;
text-align: center;
}
.shares-balance-cell {
text-style: bold;
width: 3fr;
text-align: right;
}
/* Rows of the table */
.odd-row {
background: $row-color-1;
}
.even-row {
background: $row-color-2;
}
from __future__ import annotations
from typing import TYPE_CHECKING
from textual.widgets import TabPane
from clive.__private.ui.widgets.clive_widget import CliveWidget
if TYPE_CHECKING:
from rich.text import TextType
class PowerDown(TabPane, CliveWidget):
"""TabPane with all content about power down."""
def __init__(self, title: TextType):
"""
Initialize a TabPane.
Args:
----
title: Title of the TabPane (will be displayed in a tab label).
"""
super().__init__(title=title)