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: str, name: str, room: str): super().__init__(id=id, name=name, room=room, device_type="bedscale") self._ip_address = ip_address async def poll_weights(self) -> BedscaleWeightResult: loop = asyncio.get_event_loop() tr_request = loop.run_in_executor(None, self.__poll_scale__, "tr") tl_request = loop.run_in_executor(None, self.__poll_scale__, "tl") br_request = loop.run_in_executor(None, self.__poll_scale__, "br") bl_request = loop.run_in_executor(None, self.__poll_scale__, "bl") results = BedscaleWeightResult( tr=await tr_request, tl=await tl_request, br=await br_request, bl=await bl_request, ) # TODO: Sanity checks # TODO: Keep track of empty-bed weight 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