mash-server/src/bridges/bedscale/bedscale_entity.py

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/{leg}/").json()["value"]
        except:
            return None