41 lines
1.3 KiB
Python
41 lines
1.3 KiB
Python
|
import requests
|
||
|
|
import asyncio
|
||
|
|
from core.entity import Entity
|
||
|
|
|
||
|
|
|
||
|
|
class BedscaleWeightResult:
|
||
|
|
def __init__(self, *, tr: float, tl: float, br: float, bl: float):
|
||
|
|
self.top_right = tr
|
||
|
|
self.top_left = tl
|
||
|
|
self.bottom_right = br
|
||
|
|
self.bottom_left = bl
|
||
|
|
|
||
|
|
# Calculate total if all values available
|
||
|
|
self.total: float | None = None
|
||
|
|
if None not in [bl, br, tl, tr]:
|
||
|
|
self.total = sum([bl, br, tl, tr])
|
||
|
|
|
||
|
|
|
||
|
|
class BedscaleEntity(Entity):
|
||
|
|
def __init__(self, *, ip_address: str, id, name, room):
|
||
|
|
super().__init__(id=id, name=name, room=room, device_type="bedscale")
|
||
|
|
self._ip_address = ip_address
|
||
|
|
|
||
|
|
async def poll_weights(self) -> BedscaleWeightResult:
|
||
|
|
results = BedscaleWeightResult(
|
||
|
|
tr=await asyncio.run_in_executor(None, self.__poll_scale__, "tr"),
|
||
|
|
tl=await asyncio.run_in_executor(None, self.__poll_scale__, "tl"),
|
||
|
|
br=await asyncio.run_in_executor(None, self.__poll_scale__, "br"),
|
||
|
|
bl=await asyncio.run_in_executor(None, self.__poll_scale__, "bl"),
|
||
|
|
)
|
||
|
|
|
||
|
|
# TODO: Sanity checks
|
||
|
|
|
||
|
|
return results
|
||
|
|
|
||
|
|
def __poll_scale__(self, leg: str) -> float | None:
|
||
|
|
try:
|
||
|
|
return requests.get(f"{self._ip_address}/sensor/br/").json()["value"]
|
||
|
|
except:
|
||
|
|
return None
|