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 (7)
......@@ -11,7 +11,7 @@ from clive.__private.config import settings
from clive.__private.core.clive_import import get_clive
from clive.__private.core.validate_schema_field import is_schema_field_valid
from clive.__private.logger import logger
from clive.__private.storage.accounts import WorkingAccount
from clive.__private.storage.accounts import Account, WorkingAccount
from clive.__private.storage.contextual import Context
from clive.__private.validators.profile_name_validator import ProfileNameValidator
from clive.core.url import Url
......@@ -25,8 +25,6 @@ if TYPE_CHECKING:
from typing_extensions import Self
from clive.__private.storage.accounts import Account
class ProfileDataError(CliveError):
"""An error related to profile data."""
......@@ -43,6 +41,14 @@ class NoWorkingAccountError(ProfileDataError):
"""No working account is available."""
class AccountNotFoundError(ProfileDataError):
"""Raised when an account is not found in `get_account_by_name` method."""
def __init__(self, account_name: str) -> None:
super().__init__(f"Account {account_name} not found in tracked accounts (working + watched accounts)")
self.account_name = account_name
class ProfileCouldNotBeLoadedError(ProfileDataError):
"""Raised when a profile could not be loaded."""
......@@ -139,6 +145,14 @@ class ProfileData(Context):
raise ProfileInvalidNameError(name, reason=str(result.failure_descriptions))
def get_account_by_name(self, value: str | Account) -> Account:
searched_account_name = value if isinstance(value, str) else value.name
for account in self.get_tracked_accounts():
if account.name == searched_account_name:
return account
raise AccountNotFoundError(searched_account_name)
@property
def working_account(self) -> WorkingAccount:
"""
......
......@@ -6,7 +6,6 @@ from textual import on
from textual.containers import Horizontal
from textual.widgets import Static, TabPane
from clive.__private.storage.accounts import Account, WorkingAccount
from clive.__private.ui.account_details.alarms.alarm_fix_details import get_detailed_alarm_fix_details
from clive.__private.ui.account_details.alarms.alarm_info_screen.alarm_info_screen import AlarmInfoScreen
from clive.__private.ui.get_css import get_css_from_relative_path
......@@ -30,6 +29,7 @@ if TYPE_CHECKING:
from clive.__private.core.alarms.alarm import AnyAlarm
from clive.__private.core.profile_data import ProfileData
from clive.__private.core.world import TextualWorld
from clive.__private.storage.accounts import Account
from clive.__private.ui.account_details.alarms.alarm_fix_details import AlarmFixDetails
......@@ -92,26 +92,9 @@ class AlarmsTable(CliveCheckerboardTable):
account = self._get_actual_account_state(content)
self._previous_alarms = account.alarms.harmful_alarms
def _get_account_from_watched_accounts(self, content: ProfileData) -> Account:
"""Search for the account in the watched account and return matched account."""
searched_account = None
for account in content.watched_accounts:
if account == self._account:
searched_account = account
assert searched_account is not None, f"Account {self._account} not found in watched accounts"
return searched_account
def _get_actual_account_state(self, content: ProfileData) -> Account:
"""Return the account with the actual state."""
return (
content.working_account
if self.is_current_account_working
else self._get_account_from_watched_accounts(content)
)
@property
def is_current_account_working(self) -> bool:
return isinstance(self._account, WorkingAccount)
return content.get_account_by_name(self._account)
class Alarms(TabPane, CliveWidget):
......
......@@ -59,14 +59,26 @@ class WithdrawalInfo(Vertical, CliveWidget):
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
self._provider,
"_content",
self._get_next_withdrawal_date,
id_="withdrawal-info-date",
first_try_callback=lambda content: content is not None,
)
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
self._provider,
"_content",
self._get_to_withdraw_hp,
id_="withdrawal-info-vests-amount",
first_try_callback=lambda content: content is not None,
)
yield DynamicLabel(
self._provider, "_content", self._get_to_withdraw_vests, id_="withdrawal-info-hp-amount", init=False
self._provider,
"_content",
self._get_to_withdraw_vests,
id_="withdrawal-info-hp-amount",
first_try_callback=lambda content: content is not None,
)
def _get_next_withdrawal_date(self, content: HivePowerData) -> str:
......@@ -92,7 +104,12 @@ class APR(DynamicLabel):
"""
def __init__(self, provider: HivePowerDataProvider) -> None:
super().__init__(obj_to_watch=provider, attribute_name="_content", callback=self._get_apr, init=False)
super().__init__(
obj_to_watch=provider,
attribute_name="_content",
callback=self._get_apr,
first_try_callback=lambda content: content is not None,
)
self._provider = provider
def _get_apr(self, content: HivePowerData) -> str:
......
......@@ -13,7 +13,10 @@ if TYPE_CHECKING:
class HpVestsFactor(Notice):
def __init__(self, provider: HivePowerDataProvider) -> None:
super().__init__(
obj_to_watch=provider, attribute_name="_content", callback=self._get_hp_vests_factor, init=False
obj_to_watch=provider,
attribute_name="_content",
callback=self._get_hp_vests_factor,
first_try_callback=lambda content: content is not None,
)
def _get_hp_vests_factor(self, content: HivePowerData) -> str:
......
......@@ -8,6 +8,7 @@ from clive.__private.ui.widgets.dynamic_label import DynamicLabel
if TYPE_CHECKING:
from collections.abc import Callable
from clive.__private.core.profile_data import ProfileData
from clive.__private.storage.accounts import Account
......@@ -32,7 +33,10 @@ class AccountReferencingWidget(CliveWidget):
self.app.world,
"profile_data",
lambda: foo() if self._account.name else "NULL",
first_try_callback=lambda: self._account.is_node_data_available,
first_try_callback=self._check_if_account_node_data_is_available,
classes=classes,
init=init,
)
def _check_if_account_node_data_is_available(self, profile_data: ProfileData) -> bool:
return profile_data.get_account_by_name(self._account).is_node_data_available
from __future__ import annotations
from inspect import isawaitable
from typing import TYPE_CHECKING, Any, Awaitable, Callable, Union
from typing import TYPE_CHECKING, Any, Awaitable, Callable, TypeVar, cast, overload
from textual.widgets import Label
......@@ -14,16 +14,18 @@ if TYPE_CHECKING:
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],
]
T = TypeVar("T")
FirstTryCallbackType = Callable[[], bool]
WatchLikeCallbackBothValuesType = Callable[[Any, Any], Awaitable[T]] | Callable[[Any, Any], T]
WatchLikeCallbackNewValueType = Callable[[Any], Awaitable[T]] | Callable[[Any], T]
WatchLikeCallbackNoArgsType = Callable[[], Awaitable[T]] | Callable[[], T]
WatchLikeCallbackType = (
WatchLikeCallbackBothValuesType[T] | WatchLikeCallbackNewValueType[T] | WatchLikeCallbackNoArgsType[T]
)
DynamicLabelCallbackType = WatchLikeCallbackType[str]
DynamicLabelFirstTryCallbackType = WatchLikeCallbackType[bool]
class DynamicLabel(CliveWidget):
......@@ -47,7 +49,7 @@ class DynamicLabel(CliveWidget):
attribute_name: str,
callback: DynamicLabelCallbackType,
*,
first_try_callback: FirstTryCallbackType = lambda: True,
first_try_callback: DynamicLabelFirstTryCallbackType = lambda: True,
prefix: str = "",
init: bool = True,
shrink: bool = False,
......@@ -81,18 +83,43 @@ class DynamicLabel(CliveWidget):
async def attribute_changed(self, old_value: Any, value: Any) -> None: # noqa: ANN401
callback = self.__callback
if not self._first_try_callback():
should_update = self._call_with_arbitrary_args(self._first_try_callback, old_value, value)
if not should_update:
return
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]
result = self._call_with_arbitrary_args(callback, old_value, value)
if isawaitable(result):
result = await result
if result != self.__label.renderable:
self.__label.update(f"{self.__prefix}{result}")
@overload
def _call_with_arbitrary_args(
self,
callback: DynamicLabelCallbackType,
old_value: Any, # noqa: ANN401
value: Any, # noqa: ANN401
) -> Awaitable[str] | str: ...
@overload
def _call_with_arbitrary_args(
self,
callback: DynamicLabelFirstTryCallbackType,
old_value: Any, # noqa: ANN401
value: Any, # noqa: ANN401
) -> Awaitable[bool] | bool: ...
def _call_with_arbitrary_args(
self,
callback: DynamicLabelCallbackType | DynamicLabelFirstTryCallbackType,
old_value: Any,
value: Any,
) -> Awaitable[str] | str | Awaitable[bool] | bool:
param_count = count_parameters(callback)
if param_count == 2: # noqa: PLR2004
return cast(WatchLikeCallbackBothValuesType[Any], callback)(old_value, value)
if param_count == 1:
return cast(WatchLikeCallbackNewValueType[Any], callback)(old_value)
return cast(WatchLikeCallbackNoArgsType[Any], callback)()
......@@ -75,8 +75,8 @@ class AlarmDisplay(DynamicLabel):
self.app.world,
"profile_data",
update_callback,
first_try_callback=lambda: all(
acc.is_alarms_data_available for acc in account_getter(self.app.world.profile_data)
first_try_callback=lambda profile_data: all(
acc.is_alarms_data_available for acc in account_getter(profile_data)
),
id_=id_,
classes=classes,
......
......@@ -7,7 +7,7 @@ from clive.__private.ui.widgets.titled_label import TitledLabel
if TYPE_CHECKING:
from textual.reactive import Reactable
from clive.__private.ui.widgets.dynamic_label import DynamicLabelCallbackType
from clive.__private.ui.widgets.dynamic_label import DynamicLabelCallbackType, DynamicLabelFirstTryCallbackType
class Notice(TitledLabel):
......@@ -28,6 +28,7 @@ class Notice(TitledLabel):
obj_to_watch: Reactable | None = None,
attribute_name: str | None = None,
callback: DynamicLabelCallbackType | None = None,
first_try_callback: DynamicLabelFirstTryCallbackType = lambda: True,
init: bool = True,
) -> None:
super().__init__(
......@@ -36,5 +37,6 @@ class Notice(TitledLabel):
obj_to_watch=obj_to_watch,
attribute_name=attribute_name,
callback=callback,
first_try_callback=first_try_callback,
init=init,
)
......@@ -6,7 +6,11 @@ from textual.widgets import Label
from clive.__private.ui.get_css import get_css_from_relative_path
from clive.__private.ui.widgets.clive_widget import CliveWidget
from clive.__private.ui.widgets.dynamic_label import DynamicLabel, DynamicLabelCallbackType
from clive.__private.ui.widgets.dynamic_label import (
DynamicLabel,
DynamicLabelCallbackType,
DynamicLabelFirstTryCallbackType,
)
if TYPE_CHECKING:
from rich.console import RenderableType
......@@ -32,6 +36,7 @@ class TitledLabel(CliveWidget):
obj_to_watch: Reactable | None = None,
attribute_name: str | None = None,
callback: DynamicLabelCallbackType | None = None,
first_try_callback: DynamicLabelFirstTryCallbackType = lambda: True,
init: bool = True,
id_: str | None = None,
) -> None:
......@@ -41,7 +46,13 @@ class TitledLabel(CliveWidget):
self.__value_label: DynamicLabel | Label = (
DynamicLabel(
obj_to_watch, attribute_name, callback, prefix=self.__formatted_value(), init=init, id_="value"
obj_to_watch,
attribute_name,
callback,
prefix=self.__formatted_value(),
first_try_callback=first_try_callback,
init=init,
id_="value",
)
if obj_to_watch and attribute_name and callback
else Label(self.__formatted_value(), id="value")
......