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 (14)
Showing
with 933 additions and 49 deletions
......@@ -9,7 +9,7 @@ if TYPE_CHECKING:
@lru_cache(maxsize=2048)
def __count_parameters(callback: Callable[..., Any]) -> int:
def count_parameters(callback: Callable[..., Any]) -> int:
"""Count the number of parameters in a callable."""
return len(signature(callback).parameters)
......@@ -27,7 +27,7 @@ async def invoke(callback: Callable[..., Any], *params: Any) -> Any:
-------
The return value of the invoked callable.
"""
parameter_count = __count_parameters(callback)
parameter_count = count_parameters(callback)
result = callback(*params[:parameter_count])
if isawaitable(result):
result = await result
......
......@@ -12,6 +12,7 @@ from textual import on, work
from textual._context import active_message_pump
from textual.app import App, AutopilotCallbackType
from textual.binding import Binding
from textual.events import ScreenResume
from textual.notifications import Notification, Notify, SeverityLevel
from textual.reactive import reactive, var
......@@ -262,7 +263,9 @@ class Clive(App[int], ManualReactive):
with self.batch_update():
while not self.__screen_eq(self.screen_stack[-1], screen):
self.pop_screen()
with self.prevent(ScreenResume):
self.pop_screen()
self.screen.post_message(ScreenResume())
break # Screen found and located on top of the stack, stop
else:
raise ScreenNotFoundError(
......
......@@ -101,12 +101,12 @@ class AccountInfo(Container, AccountReferencingWidget):
yield DynamicLabel(
self.app.world,
"profile_data",
lambda _: f"History entry: {humanize_datetime(self._account.data.last_history_entry)}",
lambda: f"History entry: {humanize_datetime(self._account.data.last_history_entry)}",
)
yield DynamicLabel(
self.app.world,
"profile_data",
lambda _: f"Account update: {humanize_datetime(self._account.data.last_account_update)}",
lambda: f"Account update: {humanize_datetime(self._account.data.last_account_update)}",
)
......
......@@ -3,7 +3,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING
from clive.__private.core.hive_vests_conversions import hive_to_vests
from clive.__private.ui.widgets.dynamic_label import DynamicLabel
from clive.__private.ui.widgets.notice import Notice
from clive.models import Asset
if TYPE_CHECKING:
......@@ -11,23 +11,12 @@ if TYPE_CHECKING:
from clive.__private.ui.data_providers.hive_power_data_provider import HivePowerDataProvider
class HpVestsFactor(DynamicLabel):
DEFAULT_CSS = """
HpVestsFactor {
width: 1fr;
height: 1;
margin-bottom: 1;
align: center middle;
background: $warning;
color: $text;
}
"""
class HpVestsFactor(Notice):
def __init__(self, provider: HivePowerDataProvider):
super().__init__(
obj_to_watch=provider, attribute_name="_content", callback=self._get_hp_vests_factor, init=False
)
def _get_hp_vests_factor(self, content: HivePowerData) -> str:
factor = hive_to_vests(1000, content.gdpo)
factor = hive_to_vests(Asset.hive(1), content.gdpo)
return f"HP is calculated to VESTS with the factor: 1.000 HP -> {Asset.pretty_amount(factor)} VESTS"
from __future__ import annotations
from textual.widgets import Static
class OperationNameInfo(Static):
"""Widget used to inform the user about the real name of the operation in the blockchain."""
DEFAULT_CSS = """
OperationNameInfo {
text-style: bold;
margin-bottom: 1;
background: $accent;
width: 1fr;
height: 1;
text-align: center;
}
"""
def __init__(self, operation_name: str, operation_alias: str):
"""
Initialize the `OperationNameInfo` widget.
Args:
----
operation_name: the real name of the operation in the blockchain.
operation_alias: common name of the operation.
"""
super().__init__(
renderable=f"{operation_alias.capitalize()} corresponds to a `{operation_name.lower()}` blockchain operation"
)
......@@ -23,6 +23,7 @@ from clive.__private.ui.operations.hive_power_management.delegate_hive_power.del
)
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.hive_power_management.withdraw_routes.withdraw_routes import WithdrawRoutes
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
......@@ -33,6 +34,7 @@ if TYPE_CHECKING:
POWER_UP_TAB_LABEL: Final[str] = "Power up"
POWER_DOWN_TAB_LABEL: Final[str] = "Power down"
WITHDRAW_ROUTES_TAB_LABEL: Final[str] = "Withdraw routes"
DELEGATE_HIVE_POWER_LABEL: Final[str] = "Delegate"
......@@ -59,4 +61,5 @@ class HivePowerManagement(OperationBaseScreen):
with CliveTabbedContent():
yield PowerUp(POWER_UP_TAB_LABEL)
yield PowerDown(POWER_DOWN_TAB_LABEL)
yield WithdrawRoutes(WITHDRAW_ROUTES_TAB_LABEL)
yield DelegateHivePower(DELEGATE_HIVE_POWER_LABEL)
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING
from textual.widgets import TabPane
from textual import on
from textual.containers import Horizontal
from textual.widgets import Pretty, Static, TabPane
from clive.__private.core.formatters.humanize import humanize_datetime
from clive.__private.core.hive_vests_conversions import hive_to_vests
from clive.__private.ui.data_providers.hive_power_data_provider import HivePowerDataProvider
from clive.__private.ui.get_css import get_css_from_relative_path
from clive.__private.ui.operations.bindings.operation_action_bindings import OperationActionBindings
from clive.__private.ui.operations.hive_power_management.common_hive_power.hp_vests_factor import HpVestsFactor
from clive.__private.ui.operations.hive_power_management.common_hive_power.operation_name_widget import (
OperationNameInfo,
)
from clive.__private.ui.operations.operation_summary.cancel_power_down import CancelPowerDown
from clive.__private.ui.widgets.can_focus_with_scrollbars_only import CanFocusWithScrollbarsOnly
from clive.__private.ui.widgets.clive_button import CliveButton
from clive.__private.ui.widgets.clive_checkerboard_table import (
EVEN_CLASS_NAME,
ODD_CLASS_NAME,
CliveCheckerboardTable,
CliveCheckerBoardTableCell,
CliveCheckerboardTableRow,
)
from clive.__private.ui.widgets.clive_widget import CliveWidget
from clive.__private.ui.widgets.currency_selector.currency_selector_hp_vests import CurrencySelectorHpVests
from clive.__private.ui.widgets.generous_button import GenerousButton
from clive.__private.ui.widgets.inputs.hp_vests_amount_input import HPVestsAmountInput
from clive.__private.ui.widgets.notice import Notice
from clive.models import Asset
from schemas.operations import WithdrawVestingOperation
if TYPE_CHECKING:
from rich.text import TextType
from textual.app import ComposeResult
from textual.widget import Widget
from clive.__private.core.commands.data_retrieval.hive_power_data import HivePowerData
class PowerDown(TabPane, CliveWidget):
class PlaceTaker(Static):
pass
class ScrollablePart(CanFocusWithScrollbarsOnly):
pass
class WithdrawRoutesDisplay(CliveWidget):
"""Widget used just to inform user to which account has withdrawal route and how much % it is."""
def compose(self) -> ComposeResult:
yield Static("Loading...", id="withdraw-routes-header")
yield Pretty({})
def on_mount(self) -> None:
self.watch(self.provider, "_content", self._update_withdraw_routes, init=False)
def _update_withdraw_routes(self, content: HivePowerData) -> None:
"""Update withdraw routes pretty widget."""
if not content.withdraw_routes:
self.query_one("#withdraw-routes-header", Static).update("You have no withdraw routes")
self.query_one(Pretty).display = False
return
withdraw_routes = {}
for withdraw_route in content.withdraw_routes:
withdraw_routes[withdraw_route.to_account] = f"{withdraw_route.percent / 100}%"
self.query_one("#withdraw-routes-header", Static).update("Your withdraw routes")
self.query_one(Pretty).update(withdraw_routes)
self.display = True
@property
def provider(self) -> HivePowerDataProvider:
return self.screen.query_one(HivePowerDataProvider)
class PendingPowerDownHeader(Horizontal):
def compose(self) -> ComposeResult:
yield Static("Next power down", classes=ODD_CLASS_NAME)
yield Static("Power down [HP]", classes=EVEN_CLASS_NAME)
yield Static("Power down [VESTS]", classes=ODD_CLASS_NAME)
yield PlaceTaker()
class PendingPowerDown(CliveCheckerboardTable):
def __init__(self) -> None:
super().__init__(
Static("Current power down", id="current-power-down-title"), PendingPowerDownHeader(), dynamic=True
)
self._previous_next_vesting_withdrawal: datetime = datetime.min
def create_dynamic_rows(self, content: HivePowerData) -> list[CliveCheckerboardTableRow]:
self._previous_next_vesting_withdrawal = content.next_vesting_withdrawal
return [
CliveCheckerboardTableRow(
CliveCheckerBoardTableCell(humanize_datetime(content.next_vesting_withdrawal)),
CliveCheckerBoardTableCell(Asset.pretty_amount(content.next_power_down.hp_balance)),
CliveCheckerBoardTableCell(Asset.pretty_amount(content.next_power_down.vests_balance)),
CliveCheckerBoardTableCell(CliveButton("Cancel", variant="error")),
)
]
def get_no_content_available_widget(self) -> Widget:
return Static("You have no current power down process", id="no-current-power-down-info")
@on(CliveButton.Pressed)
def push_operation_summary_screen(self) -> None:
self.app.push_screen(
CancelPowerDown(self.provider.content.next_vesting_withdrawal, self.provider.content.next_power_down)
)
@property
def is_anything_to_display(self) -> bool:
return humanize_datetime(self.provider.content.next_vesting_withdrawal) != "never"
@property
def provider(self) -> HivePowerDataProvider:
return self.screen.query_one(HivePowerDataProvider)
@property
def check_if_should_be_updated(self) -> bool:
return self.provider.content.next_vesting_withdrawal != self._previous_next_vesting_withdrawal
class PowerDown(TabPane, OperationActionBindings):
"""TabPane with all content about power down."""
def __init__(self, title: TextType):
DEFAULT_CSS = get_css_from_relative_path(__file__)
def __init__(self, title: TextType) -> None:
"""
Initialize a TabPane.
Initialize the PowerDown tab-pane.
Args:
----
title: Title of the TabPane (will be displayed in a tab label).
"""
super().__init__(title=title)
self._shares_input = HPVestsAmountInput()
self._one_withdrawal_display = Notice(
obj_to_watch=self._shares_input.input,
attribute_name="value",
callback=self._calculate_one_withdrawal,
init=False,
)
self._one_withdrawal_display.display = False
def compose(self) -> ComposeResult:
with ScrollablePart():
yield OperationNameInfo("withdraw vesting", "power down")
yield HpVestsFactor(self.provider)
with Horizontal(id="input-with-button"):
yield self._shares_input
yield GenerousButton(self._shares_input, self._get_shares_balance) # type: ignore[arg-type]
yield self._one_withdrawal_display
yield PendingPowerDown()
yield WithdrawRoutesDisplay()
def _get_shares_balance(self) -> Asset.Hive | Asset.Vests:
if self._shares_input.selected_asset_type is Asset.Hive:
return self.provider.content.owned_balance.hp_balance - self.provider.content.delegated_balance.hp_balance
return self.provider.content.owned_balance.vests_balance - self.provider.content.delegated_balance.vests_balance
def _calculate_one_withdrawal(self) -> str:
"""The withdrawal is divided into 13 parts - calculate and inform the user of the amount of one of them."""
shares_input = self._shares_input.value_or_none()
if shares_input is None:
self._one_withdrawal_display.display = False
return ""
one_withdrawal = shares_input / 13
self._one_withdrawal_display.display = True
asset = f"{Asset.pretty_amount(one_withdrawal)} {'VESTS' if isinstance(one_withdrawal, Asset.Vests) else 'HP'}"
return f"The withdrawal will be divided into 13 parts, one of which is: {asset}"
@on(CurrencySelectorHpVests.Changed)
def shares_type_changed(self) -> None:
"""Clear input when shares type was changed and hide factor display when vests selected."""
self._shares_input.input.clear()
hp_vests_factor = self.query_one(HpVestsFactor)
if self._shares_input.selected_asset_type is Asset.Vests:
hp_vests_factor.display = False
return
hp_vests_factor.display = True
def _create_operation(self) -> WithdrawVestingOperation | None:
asset = self._shares_input.value_or_none()
if asset is None:
return None
if isinstance(asset, Asset.Hive):
# If the user has passed an amount in `HP` - convert it to `VESTS`. The operation is performed using VESTS.
asset = hive_to_vests(asset, self.provider.content.gdpo)
return WithdrawVestingOperation(account=self.working_account, vesting_shares=asset)
@property
def provider(self) -> HivePowerDataProvider:
return self.screen.query_one(HivePowerDataProvider)
@property
def working_account(self) -> str:
return self.app.world.profile_data.working_account.name
CliveButton {
width: 1fr;
}
ScrollablePart {
height: auto;
}
WithdrawRoutesDisplay Pretty {
content-align: center middle;
width: 1fr;
}
WithdrawRoutesDisplay {
height: auto;
}
HpVestsFactor {
margin-bottom: 1;
}
#input-with-button {
background: $panel;
padding: 2 2;
height: auto;
}
#withdraw-routes-header {
text-style: bold;
margin-top: 1;
background: $primary;
width: 1fr;
height: 1;
text-align: center;
}
/* Pending power down */
#current-power-down-title {
text-style: bold;
margin-top: 1;
background: $primary;
width: 1fr;
height: 1;
text-align: center;
}
#no-current-power-down-info {
text-style: bold;
margin-top: 1;
background: $primary;
width: 1fr;
height: 1;
text-align: center;
}
PendingPowerDown {
height: auto;
}
PendingPowerDownHeader {
height: 1;
}
PendingPowerDownHeader Static {
text-style: bold;
width: 1fr;
text-align: center;
}
......@@ -2,11 +2,15 @@ from __future__ import annotations
from typing import TYPE_CHECKING
from textual.containers import Horizontal, ScrollableContainer, Vertical
from textual.widgets import Static, TabPane
from textual.containers import Horizontal, Vertical
from textual.widgets import TabPane
from clive.__private.ui.get_css import get_css_from_relative_path
from clive.__private.ui.operations.bindings.operation_action_bindings import OperationActionBindings
from clive.__private.ui.operations.hive_power_management.common_hive_power.operation_name_widget import (
OperationNameInfo,
)
from clive.__private.ui.widgets.can_focus_with_scrollbars_only import CanFocusWithScrollbarsOnly
from clive.__private.ui.widgets.generous_button import GenerousButton
from clive.__private.ui.widgets.inputs.account_name_input import AccountNameInput
from clive.__private.ui.widgets.inputs.clive_validated_input import CliveValidatedInput
......@@ -21,7 +25,7 @@ if TYPE_CHECKING:
from clive.models import Asset
class ScrollablePart(ScrollableContainer):
class ScrollablePart(CanFocusWithScrollbarsOnly):
pass
......@@ -44,7 +48,7 @@ class PowerUp(TabPane, OperationActionBindings):
def compose(self) -> ComposeResult:
with ScrollablePart():
yield Static("Power up corresponds to a `transfer to vesting` operation", id="operation-name-info")
yield OperationNameInfo("transfer to vesting", "power up")
yield Notice("Your governance voting power will be increased after 30 days")
with Vertical(id="power-up-inputs"):
yield self._receiver_input
......
......@@ -2,6 +2,10 @@ PowerUp {
height: auto;
}
ScrollablePart {
height: auto;
}
Notice {
margin-top: 1;
margin-bottom: 1;
......@@ -23,12 +27,3 @@ Notice {
margin-top: 1;
height: auto;
}
#input-with-button HiveAssetAmountInput {
width: 1fr;
}
#input-with-button GenerousButton {
width: 14;
min-width: 14;
}
from __future__ import annotations
from typing import TYPE_CHECKING
from textual import on
from textual.containers import Horizontal, ScrollableContainer, Vertical
from textual.widgets import Checkbox, Static, TabPane
from clive.__private.ui.data_providers.hive_power_data_provider import HivePowerDataProvider
from clive.__private.ui.get_css import get_css_from_relative_path
from clive.__private.ui.operations.bindings import OperationActionBindings
from clive.__private.ui.operations.operation_summary.remove_withdraw_vesting_route import RemoveWithdrawVestingRoute
from clive.__private.ui.widgets.clive_button import CliveButton
from clive.__private.ui.widgets.clive_checkerboard_table import (
EVEN_CLASS_NAME,
ODD_CLASS_NAME,
CliveCheckerboardTable,
CliveCheckerBoardTableCell,
CliveCheckerboardTableRow,
)
from clive.__private.ui.widgets.inputs.account_name_input import AccountNameInput
from clive.__private.ui.widgets.inputs.clive_validated_input import CliveValidatedInput
from clive.__private.ui.widgets.inputs.percent_input import PercentInput
from schemas.operations import SetWithdrawVestingRouteOperation
if TYPE_CHECKING:
from rich.text import TextType
from textual.app import ComposeResult
from textual.widget import Widget
from clive.__private.core.commands.data_retrieval.hive_power_data import HivePowerData
from schemas.apis.database_api.fundaments_of_reponses import WithdrawVestingRoutesFundament as WithdrawRouteSchema
class PlaceTaker(Static):
pass
class WithdrawRoutesHeader(Horizontal):
def compose(self) -> ComposeResult:
yield Static("To", classes=EVEN_CLASS_NAME)
yield Static("Percent", classes=ODD_CLASS_NAME)
yield Static("Auto vest", classes=EVEN_CLASS_NAME)
yield PlaceTaker()
class WithdrawRoute(CliveCheckerboardTableRow):
"""Row of the `WithdrawRoutesTable`."""
def __init__(self, withdraw_route: WithdrawRouteSchema) -> None:
super().__init__(
CliveCheckerBoardTableCell(withdraw_route.to_account),
CliveCheckerBoardTableCell(f"{withdraw_route.percent / 100} %"),
CliveCheckerBoardTableCell(f"{withdraw_route.auto_vest}"),
CliveCheckerBoardTableCell(CliveButton("Remove", classes="remove-withdraw-route-button", variant="error")),
)
self._withdraw_route = withdraw_route
@on(CliveButton.Pressed, ".remove-withdraw-route-button")
def push_operation_summary_screen(self) -> None:
self.app.push_screen(RemoveWithdrawVestingRoute(self._withdraw_route))
class WithdrawRoutesTable(CliveCheckerboardTable):
"""Table with WithdrawRoutes."""
def __init__(self) -> None:
super().__init__(
Static("Current withdraw routes", id="withdraw-routes-table-title"), WithdrawRoutesHeader(), dynamic=True
)
self._withdraw_routes: list[WithdrawRouteSchema] | None = None
def create_dynamic_rows(self, content: HivePowerData) -> list[CliveCheckerboardTableRow]:
self._withdraw_routes = content.withdraw_routes
withdraw_routes: list[CliveCheckerboardTableRow] = [
WithdrawRoute(withdraw_route) for withdraw_route in self._withdraw_routes
]
return withdraw_routes
def get_no_content_available_widget(self) -> Widget:
return Static("You have no withdraw routes", id="no-withdraw-routes-info")
@property
def provider(self) -> HivePowerDataProvider:
return self.screen.query_one(HivePowerDataProvider)
@property
def check_if_should_be_updated(self) -> bool:
return self.provider.content.withdraw_routes != self._withdraw_routes
@property
def is_anything_to_display(self) -> bool:
return len(self.provider.content.withdraw_routes) != 0
@property
def working_account(self) -> str:
return self.app.world.profile_data.working_account.name
class WithdrawRoutes(TabPane, OperationActionBindings):
"""TabPane with all content about setting withdraw routes."""
DEFAULT_CSS = get_css_from_relative_path(__file__)
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)
self._account_input = AccountNameInput()
self._percent_input = PercentInput("Percent")
self._auto_vest_checkbox = Checkbox("Auto vest")
def compose(self) -> ComposeResult:
with ScrollableContainer():
yield Static("Set withdraw route", id="set-withdraw-route-title")
with Vertical(id="inputs-container"):
yield self._account_input
with Horizontal(id="input-with-checkbox"):
yield self._percent_input
yield self._auto_vest_checkbox
yield WithdrawRoutesTable()
def _create_operation(self) -> SetWithdrawVestingRouteOperation | None:
CliveValidatedInput.validate_many(self._account_input, self._percent_input)
return SetWithdrawVestingRouteOperation(
from_account=self.working_account,
to_account=self._account_input.value_or_error,
percent=self._percent_input.value_or_error * 100,
auto_vest=self._auto_vest_checkbox.value,
)
@property
def working_account(self) -> str:
return self.app.world.profile_data.working_account.name
WithdrawRoutes {
height: 1fr;
}
WithdrawRoutes Static {
text-style: bold;
width: 1fr;
text-align: center;
}
/* Set withdraw route */
#set-withdraw-route-title {
background: $primary;
width: 1fr;
height: 1;
}
#inputs-container {
background: $panel;
padding: 2 4;
height: auto;
}
#input-with-checkbox {
margin-top: 1;
height: auto;
}
#input-with-checkbox IntegerInput {
width: 1fr;
}
#input-with-checkbox Checkbox {
width: 17;
}
AccountNameInput KnownAccount {
margin-left: 3;
}
/* Withdraw routes table */
WithdrawRoutesHeader {
height: auto;
}
WithdrawRoutesTable {
margin-top: 1;
height: auto;
}
#no-withdraw-routes-info {
background: $primary;
height: auto;
}
#withdraw-routes-table-title {
background: $primary;
width: 1fr;
}
WithdrawRoute {
height: auto;
}
WithdrawRoute .remove-withdraw-route-button {
background: $error-darken-1;
width: 1fr;
}
from __future__ import annotations
from typing import TYPE_CHECKING, ClassVar
from clive.__private.core.formatters.humanize import humanize_datetime
from clive.__private.ui.operations.operation_summary.operation_summary import OperationSummary
from clive.__private.ui.widgets.inputs.labelized_input import LabelizedInput
from clive.models import Asset
from schemas.operations import WithdrawVestingOperation
if TYPE_CHECKING:
from datetime import datetime
from textual.app import ComposeResult
from clive.__private.core.commands.data_retrieval.hive_power_data import SharesBalance
class CancelPowerDown(OperationSummary):
BIG_TITLE: ClassVar[str] = "Cancel power down"
def __init__(self, next_power_down_date: datetime, next_power_down: SharesBalance) -> None:
super().__init__()
self._next_power_down_date = next_power_down_date
self._next_power_down = next_power_down
def content(self) -> ComposeResult:
yield LabelizedInput("Next power down", humanize_datetime(self._next_power_down_date))
yield LabelizedInput("Power down [HP]", Asset.pretty_amount(self._next_power_down.hp_balance))
yield LabelizedInput("Power down [VESTS]", Asset.pretty_amount(self._next_power_down.vests_balance))
def _create_operation(self) -> WithdrawVestingOperation:
return WithdrawVestingOperation(
account=self.working_account,
vesting_shares=Asset.vests(0),
)
@property
def working_account(self) -> str:
return self.app.world.profile_data.working_account.name
from __future__ import annotations
from typing import TYPE_CHECKING, ClassVar, Final
from clive.__private.ui.operations.operation_summary.operation_summary import OperationSummary
from clive.__private.ui.widgets.inputs.labelized_input import LabelizedInput
from schemas.operations import SetWithdrawVestingRouteOperation
if TYPE_CHECKING:
from textual.app import ComposeResult
from schemas.apis.database_api.fundaments_of_reponses import WithdrawVestingRoutesFundament as WithdrawRoute
WITHDRAW_ROUTE_REMOVE_PERCENT: Final[int] = 0
class RemoveWithdrawVestingRoute(OperationSummary):
"""Screen to remove withdraw vesting route."""
BIG_TITLE: ClassVar[str] = "Remove withdraw route"
def __init__(self, withdraw_route: WithdrawRoute) -> None:
super().__init__()
self._withdraw_route = withdraw_route
def content(self) -> ComposeResult:
yield LabelizedInput("From account", self.working_account)
yield LabelizedInput("To account", self._withdraw_route.to_account)
yield LabelizedInput("Percent", str(self._withdraw_route.percent / 100))
yield LabelizedInput("Auto vest", str(self._withdraw_route.auto_vest))
def _create_operation(self) -> SetWithdrawVestingRouteOperation | None:
return SetWithdrawVestingRouteOperation(
from_account=self.working_account,
to_account=self._withdraw_route.to_account,
auto_vest=self._withdraw_route.auto_vest,
percent=WITHDRAW_ROUTE_REMOVE_PERCENT,
)
@property
def working_account(self) -> str:
return self.app.world.profile_data.working_account.name
......@@ -25,6 +25,6 @@ class AccountReferencingWidget(CliveWidget):
return DynamicLabel(
self.app.world,
"profile_data",
lambda _: foo() if self._account.name else "NULL",
lambda: foo() if self._account.name else "NULL",
classes=classes,
)
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Final
from textual.containers import Vertical
from textual.widget import Widget
from textual.widgets import Static
from clive.__private.ui.widgets.clive_widget import CliveWidget
from clive.exceptions import CliveError
if TYPE_CHECKING:
from textual.app import ComposeResult
from clive.__private.ui.data_providers.abc.data_provider import DataProvider
ODD_CLASS_NAME: Final[str] = "-odd-column"
EVEN_CLASS_NAME: Final[str] = "-even-column"
class CliveCheckerboardTableError(CliveError):
pass
class InvalidDynamicDefinedError(CliveCheckerboardTableError):
_MESSAGE = """
You are trying to create a dynamic checkerboard table without overriding one of the mandatory properties or methods.
Override it or set the `dynamic` parameter to False if you want to create a static table.
"""
def __init__(self) -> None:
super().__init__(self._MESSAGE)
class InvalidStaticDefinedError(CliveCheckerboardTableError):
_MESSAGE = """
You are trying to create a static checkerboard table without overriding the mandatory `create_static_rows` methods.
Override it or set the `dynamic` parameter to True if you want to create a dynamic table.
"""
def __init__(self) -> None:
super().__init__(self._MESSAGE)
class PlaceTaker(Static):
pass
class CliveCheckerBoardTableCell(Vertical):
"""3 - row cell of the table. Use `Static` instead if you only want single row."""
DEFAULT_CSS = """
CliveCheckerBoardTableCell {
width: 1fr;
}
CliveCheckerBoardTableCell Static {
text-style: bold;
text-align: center;
width: auto;
height: auto;
content-align: center middle;
}
"""
def __init__(self, content: str | Widget, id_: str | None = None, classes: str | None = None) -> None:
"""
Initialise the checkerboard table cell.
Args:
----
content: Text to be displayed in the cell or widget to be yielded.
even: Evenness of the cell.
id_: The ID of the widget in the DOM.
classes: The CSS classes for the widget.
"""
super().__init__(id=id_, classes=classes)
self._content = content
def compose(self) -> ComposeResult:
if isinstance(self._content, Widget):
yield self._content
else:
yield Static(self._content)
class CliveCheckerboardTableRow(CliveWidget):
"""Row with checkerboard columns."""
DEFAULT_CSS = """
CliveCheckerboardTableRow {
layout: horizontal;
height: auto;
}
"""
def __init__(self, *cells: CliveCheckerBoardTableCell):
super().__init__()
self.cells = cells
def compose(self) -> ComposeResult:
yield from self.cells
class CliveCheckerboardTable(CliveWidget):
"""
Table that displays checkerboard rows.
Dynamic usage
-------------
1. Set `dynamic` to True in `__init__`
2. Override `provider` property
3. Override `check_if_should_be_updated`
4. Override (optionally) `is_anything_to_display`
5. Override `create_dynamic_rows`
Static usage
------------
1. Override `create_static_rows`
"""
DEFAULT_CSS = """
CliveCheckerboardTable {
layout: vertical;
height: auto;
Vertical {
height: auto;
}
.-odd-column {
background: $primary-background-darken-2;
}
.-even-column {
background: $primary-background-darken-1;
}
#loading-static {
text-align: center;
text-style: bold;
}
CliveCheckerBoardTableCell {
height: 3;
Static {
width: 1fr;
height: 1fr;
}
}
}
"""
def __init__(self, title: Widget, header: Widget, dynamic: bool = False):
super().__init__()
self._title = title
self._header = header
self._dynamic = dynamic
def compose(self) -> ComposeResult:
if self._dynamic:
yield Static("Loading...", id="loading-static")
else:
self._mount_static_rows()
def on_mount(self) -> None:
if self._dynamic:
self.watch(self.provider, "_content", self._mount_dynamic_rows, init=False)
def _mount_static_rows(self) -> None:
"""Mount rows created in static mode (dynamic = False)."""
rows = self.create_static_rows()
self._set_evenness_styles(rows)
self.mount_all([self._title, self._header, *rows])
def _mount_dynamic_rows(self, content: Any) -> None:
"""New rows are mounted when the data to be displayed has been changed."""
if not self._dynamic:
raise InvalidDynamicDefinedError
if not self.check_if_should_be_updated:
return
if self.is_anything_to_display:
rows = self.create_dynamic_rows(content)
self._set_evenness_styles(rows)
widgets_to_mount = [self._title, self._header, *rows]
else:
widgets_to_mount = [self.get_no_content_available_widget()]
with self.app.batch_update():
self.query("*").remove()
self.mount_all(widgets_to_mount)
def create_dynamic_rows(self, content: Any) -> list[CliveCheckerboardTableRow]: # noqa: ARG002
"""
Override if dynamic is set to True.
Raises
------
InvalidDynamicDefinedError: When dynamic has been set to `True` without overriding the method.
"""
if self._dynamic:
raise InvalidDynamicDefinedError
return [CliveCheckerboardTableRow(CliveCheckerBoardTableCell("Define `create_dynamic_rows` method!"))]
def create_static_rows(self) -> list[CliveCheckerboardTableRow]:
"""
Override if dynamic is set to False.
Raises
------
InvalidStaticDefinedError: When dynamic has been set to `False` without overriding the method.
"""
if not self._dynamic:
raise InvalidStaticDefinedError
return [CliveCheckerboardTableRow(CliveCheckerBoardTableCell("Define `create_static_rows` method!"))]
def get_no_content_available_widget(self) -> Widget:
return Static("No content available")
@property
def check_if_should_be_updated(self) -> bool:
"""
Must be overridden by the child class when using dynamic table.
Examples
--------
return self.provider.content.actual_value != self.previous_value
Raises
------
InvalidDynamicDefinedError: When dynamic has been set to `True` without overriding the property.
"""
if self._dynamic:
raise InvalidDynamicDefinedError
return True
def _set_evenness_styles(self, rows: list[CliveCheckerboardTableRow]) -> None:
for row_index, row in enumerate(rows):
for cell_index, cell in enumerate(row.cells):
if not isinstance(cell, CliveCheckerBoardTableCell):
continue
is_even_row = row_index % 2 == 0
is_even_cell = cell_index % 2 == 0
if (is_even_row and is_even_cell) or (not is_even_row and not is_even_cell):
cell.add_class(EVEN_CLASS_NAME)
else:
cell.add_class(ODD_CLASS_NAME)
@property
def provider(self) -> DataProvider[Any]: # type: ignore[return]
"""
Must be overridden by the child class when using dynamic table.
Raises
------
InvalidDynamicDefinedError: When dynamic has been set to `True` without overriding the property.
"""
if self._dynamic:
raise InvalidDynamicDefinedError
@property
def is_anything_to_display(self) -> bool:
"""A property that checks whether there are elements to display. Should be overridden to create a custom condition."""
return True
from __future__ import annotations
from .currency_selector import CurrencySelector
from .currency_selector_base import CurrencySelectorBase
from .currency_selector_hive import CurrencySelectorHive
from .currency_selector_hp_vests import CurrencySelectorHpVests
from .currency_selector_liquid import CurrencySelectorLiquid
__all__ = [
"CurrencySelector",
"CurrencySelectorLiquid",
"CurrencySelectorHive",
"CurrencySelectorHpVests",
"CurrencySelectorBase",
]
from __future__ import annotations
from clive.__private.ui.widgets.currency_selector.currency_selector_base import (
CurrencySelectorBase,
)
from clive.models import Asset
from clive.models.asset import AssetFactoryHolder
class CurrencySelectorHive(CurrencySelectorBase[Asset.Hive]):
@staticmethod
def _create_selectable() -> dict[str, AssetFactoryHolder[Asset.Hive]]:
return {
"HIVE": AssetFactoryHolder(asset_cls=Asset.Hive, asset_factory=Asset.hive),
}
def on_mount(self) -> None:
self.disabled = True
from __future__ import annotations
from inspect import isawaitable
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, Awaitable, Callable, Union
from textual.widgets import Label
from clive.__private.core.callback import count_parameters
from clive.__private.ui.widgets.clive_widget import CliveWidget
if TYPE_CHECKING:
from collections.abc import Callable
from rich.console import RenderableType
from textual.app import ComposeResult
from textual.reactive import Reactable
DynamicLabelCallbackType = Union[
Callable[[], Awaitable[str]],
Callable[[Any], Awaitable[str]],
Callable[[Any, Any], Awaitable[str]],
Callable[[], str],
Callable[[Any], str],
Callable[[Any, Any], str],
]
class DynamicLabel(CliveWidget):
"""A label that can be updated dynamically when a reactive variable changes."""
......@@ -34,7 +43,7 @@ class DynamicLabel(CliveWidget):
self,
obj_to_watch: Reactable,
attribute_name: str,
callback: Callable[[Any], Any],
callback: DynamicLabelCallbackType,
*,
prefix: str = "",
init: bool = True,
......@@ -57,17 +66,26 @@ class DynamicLabel(CliveWidget):
return self.__label.renderable
def on_mount(self) -> None:
def delegate_work(attribute: Any) -> None:
self.run_worker(self.attribute_changed(attribute))
def delegate_work(old_value: Any, value: Any) -> None:
self.run_worker(self.attribute_changed(old_value, value))
self.watch(self.__obj_to_watch, self.__attribute_name, delegate_work, self._init)
def compose(self) -> ComposeResult:
yield self.__label
async def attribute_changed(self, attribute: Any) -> None:
value = self.__callback(attribute)
if isawaitable(value):
value = await value
if value != self.__label.renderable:
self.__label.update(f"{self.__prefix}{value}")
async def attribute_changed(self, old_value: Any, value: Any) -> None:
callback = self.__callback
param_count = count_parameters(callback)
if param_count == 2: # noqa: PLR2004
result = callback(old_value, value) # type: ignore[call-arg]
elif param_count == 1:
result = callback(value) # type: ignore[call-arg]
else:
result = callback() # type: ignore[call-arg]
if isawaitable(result):
result = await result
if result != self.__label.renderable:
self.__label.update(f"{self.__prefix}{result}")