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
        values = [bl, br, tl, tr]
        if None not in values:
            self.total = sum(values)


class BedscaleEntity(Entity):

    def __init__(
        self,
        *,
        ip_address: str,
        id: str,
        name: str,
        room: str,
        history_length: int = 60 * 5,
    ):
        super().__init__(id=id, name=name, room=room, device_type="bedscale")
        self._ip_address = ip_address
        self._history: list[BedscaleWeightResult] = []
        self._history_length = history_length
        self.latest_weight: BedscaleWeightResult = None

    async def update(self):
        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")

        new_result = 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

        self._history.append(new_result)
        if len(self._history) > self._history_length:
            self._history = self._history[-self._history_length :]

    def __poll_scale__(self, leg: str) -> float | None:
        try:
            return requests.get(f"{self._ip_address}/sensor/{leg}/").json()["value"]
        except:
            return None

    def get_history(self) -> list[BedscaleWeightResult]:
        return self._history