Lock attributes
This commit is contained in:
parent
f6139db0b5
commit
1ca89995a2
@ -330,7 +330,7 @@ class HonAppliance:
|
|||||||
command: HonCommand = self.commands.get(command_name)
|
command: HonCommand = self.commands.get(command_name)
|
||||||
for key, value in self.attributes.get("parameters", {}).items():
|
for key, value in self.attributes.get("parameters", {}).items():
|
||||||
if isinstance(value, str) and (new := command.parameters.get(key)):
|
if isinstance(value, str) and (new := command.parameters.get(key)):
|
||||||
self.attributes["parameters"][key].value = str(new.intern_value)
|
self.attributes["parameters"][key].update(str(new.intern_value), shield=True)
|
||||||
|
|
||||||
def sync_command(self, main, target=None) -> None:
|
def sync_command(self, main, target=None) -> None:
|
||||||
base: HonCommand = self.commands.get(main)
|
base: HonCommand = self.commands.get(main)
|
||||||
|
@ -1,17 +1,21 @@
|
|||||||
from datetime import datetime
|
from datetime import datetime, timedelta
|
||||||
from typing import Optional
|
from typing import Optional, Final, Dict
|
||||||
|
|
||||||
from pyhon.helper import str_to_float
|
from pyhon.helper import str_to_float
|
||||||
|
|
||||||
|
|
||||||
class HonAttribute:
|
class HonAttribute:
|
||||||
|
_LOCK_TIMEOUT: Final = 10
|
||||||
|
|
||||||
def __init__(self, data):
|
def __init__(self, data):
|
||||||
self._value: str = ""
|
self._value: str = ""
|
||||||
self._last_update: Optional[datetime] = None
|
self._last_update: Optional[datetime] = None
|
||||||
|
self._lock_timestamp: Optional[datetime] = None
|
||||||
self.update(data)
|
self.update(data)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def value(self) -> float | str:
|
def value(self) -> float | str:
|
||||||
|
"""Attribute value"""
|
||||||
try:
|
try:
|
||||||
return str_to_float(self._value)
|
return str_to_float(self._value)
|
||||||
except ValueError:
|
except ValueError:
|
||||||
@ -23,15 +27,32 @@ class HonAttribute:
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def last_update(self) -> Optional[datetime]:
|
def last_update(self) -> Optional[datetime]:
|
||||||
|
"""Timestamp of last api update"""
|
||||||
return self._last_update
|
return self._last_update
|
||||||
|
|
||||||
def update(self, data):
|
@property
|
||||||
self._value = data.get("parNewVal", "")
|
def lock(self) -> bool:
|
||||||
|
"""Shows if value changes are forbidden"""
|
||||||
|
if not self._lock_timestamp:
|
||||||
|
return False
|
||||||
|
lock_until = self._lock_timestamp + timedelta(seconds=self._LOCK_TIMEOUT)
|
||||||
|
return lock_until >= datetime.utcnow()
|
||||||
|
|
||||||
|
def update(self, data: Dict[str, str] | str, shield: bool = False) -> bool:
|
||||||
|
if self.lock and not shield:
|
||||||
|
return False
|
||||||
|
if shield:
|
||||||
|
self._lock_timestamp = datetime.utcnow()
|
||||||
|
if isinstance(data, str):
|
||||||
|
self.value = data
|
||||||
|
return True
|
||||||
|
self.value = data.get("parNewVal", "")
|
||||||
if last_update := data.get("lastUpdate"):
|
if last_update := data.get("lastUpdate"):
|
||||||
try:
|
try:
|
||||||
self._last_update = datetime.fromisoformat(last_update)
|
self._last_update = datetime.fromisoformat(last_update)
|
||||||
except ValueError:
|
except ValueError:
|
||||||
self._last_update = None
|
self._last_update = None
|
||||||
|
return True
|
||||||
|
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
return self._value
|
return self._value
|
||||||
|
Loading…
Reference in New Issue
Block a user