bambuddy/spoolbuddy/daemon/api_client.py

"""HTTP client for communicating with Bambuddy backend."""

import asyncio
import logging
from collections import deque

import httpx

logger = logging.getLogger(__name__)

MAX_BUFFER_SIZE = 100


# Where a tag write's result goes; main checks the buffer for it (#3076).
WRITE_RESULT_PATH = "/nfc/write-result"


def _says_unregistered(response: httpx.Response) -> bool:
    """The backend's own answer for a device it does not know -- not a bare
    404 from a reverse proxy while Bambuddy is restarting."""
    try:
        return response.json().get("detail") == "Device not registered"
    except Exception:
        return False


class APIClient:
    def __init__(self, backend_url: str, api_key: str):
        self._base = backend_url.rstrip("/") + "/api/v1/spoolbuddy"
        self._headers = {"X-API-Key": api_key} if api_key else {}
        self._client = httpx.AsyncClient(timeout=10.0, headers=self._headers)
        self._backoff = 1.0
        self._max_backoff = 30.0
        self._buffer: deque[dict] = deque(maxlen=MAX_BUFFER_SIZE)
        self._connected = False
        # Set when a heartbeat is refused because the backend no longer knows
        # this device (an admin unregistered it); the heartbeat loop registers
        # again and clears it.
        self.unregistered = False

    async def close(self):
        await self._client.aclose()

    async def _post(self, path: str, data: dict) -> dict | None:
        try:
            resp = await self._client.post(f"{self._base}{path}", json=data)
            resp.raise_for_status()
            self._backoff = 1.0
            self._connected = True
            return resp.json()
        except httpx.HTTPStatusError as e:
            if e.response.status_code < 500:
                # Refused, not lost: sending it again later changes nothing,
                # and a buffered refusal would block every item behind it.
                logger.warning("POST %s refused (%s)", path, e.response.status_code)
                if e.response.status_code == 404 and path.endswith("/heartbeat") and _says_unregistered(e.response):
                    self.unregistered = True
                return None
            if self._connected:
                logger.warning("Backend connection lost: %s", e)
                self._connected = False
            self._buffer.append({"path": path, "data": data})
            return None
        except Exception as e:
            if self._connected:
                logger.warning("Backend connection lost: %s", e)
                self._connected = False
            self._buffer.append({"path": path, "data": data})
            return None

    async def _get(self, path: str) -> dict | None:
        try:
            resp = await self._client.get(f"{self._base}{path}")
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            logger.warning("GET %s failed: %s", path, e)
            return None

    def has_buffered(self, path: str) -> bool:
        """Whether a POST to *path* is still waiting in the buffer to be sent."""
        return any(item["path"] == path for item in self._buffer)

    async def _flush_buffer(self):
        while self._buffer:
            item = self._buffer[0]
            try:
                resp = await self._client.post(f"{self._base}{item['path']}", json=item["data"])
                resp.raise_for_status()
                self._buffer.popleft()
            except httpx.HTTPStatusError as e:
                if e.response.status_code >= 500:
                    break
                # Refused: dropped, so the items behind it still go out.
                self._buffer.popleft()
            except Exception:
                break

    async def register_device(
        self,
        device_id: str,
        hostname: str,
        ip_address: str,
        firmware_version: str | None = None,
        has_nfc: bool = True,
        has_scale: bool = True,
        tare_offset: int = 0,
        calibration_factor: float = 1.0,
        nfc_reader_type: str | None = None,
        nfc_connection: str | None = None,
        backend_url: str | None = None,
        has_backlight: bool = False,
    ) -> dict | None:
        while True:
            result = await self._post(
                "/devices/register",
                {
                    "device_id": device_id,
                    "hostname": hostname,
                    "ip_address": ip_address,
                    "firmware_version": firmware_version,
                    "has_nfc": has_nfc,
                    "has_scale": has_scale,
                    "tare_offset": tare_offset,
                    "calibration_factor": calibration_factor,
                    "nfc_reader_type": nfc_reader_type,
                    "nfc_connection": nfc_connection,
                    "backend_url": backend_url,
                    "has_backlight": has_backlight,
                },
            )
            if result is not None:
                logger.info("Registered with backend as %s", device_id)
                return result
            logger.warning("Registration failed, retrying in %.0fs...", self._backoff)
            await asyncio.sleep(self._backoff)
            self._backoff = min(self._backoff * 2, self._max_backoff)

    async def heartbeat(
        self,
        device_id: str,
        nfc_ok: bool,
        scale_ok: bool,
        uptime_s: int,
        ip_address: str | None = None,
        firmware_version: str | None = None,
        nfc_reader_type: str | None = None,
        nfc_connection: str | None = None,
        backend_url: str | None = None,
        system_stats: dict | None = None,
    ) -> dict | None:
        payload: dict = {
            "nfc_ok": nfc_ok,
            "scale_ok": scale_ok,
            "uptime_s": uptime_s,
            "ip_address": ip_address,
            "firmware_version": firmware_version,
            "nfc_reader_type": nfc_reader_type,
            "nfc_connection": nfc_connection,
            "backend_url": backend_url,
        }
        if system_stats is not None:
            payload["system_stats"] = system_stats
        result = await self._post(
            f"/devices/{device_id}/heartbeat",
            payload,
        )
        if result and self._buffer:
            await self._flush_buffer()
        return result

    async def tag_scanned(
        self,
        device_id: str,
        tag_uid: str,
        tray_uuid: str | None = None,
        sak: int | None = None,
        tag_type: str | None = None,
        raw_blocks: dict[str, str] | None = None,
    ) -> dict | None:
        return await self._post(
            "/nfc/tag-scanned",
            {
                "device_id": device_id,
                "tag_uid": tag_uid,
                "tray_uuid": tray_uuid,
                "sak": sak,
                "tag_type": tag_type,
                "raw_blocks": raw_blocks,
            },
        )

    async def tag_removed(self, device_id: str, tag_uid: str) -> dict | None:
        return await self._post(
            "/nfc/tag-removed",
            {
                "device_id": device_id,
                "tag_uid": tag_uid,
            },
        )

    async def update_tare(self, device_id: str, tare_offset: int) -> dict | None:
        return await self._post(
            f"/devices/{device_id}/calibration/set-tare",
            {"tare_offset": tare_offset},
        )

    async def scale_reading(
        self, device_id: str, weight_grams: float, stable: bool, raw_adc: int | None = None
    ) -> dict | None:
        return await self._post(
            "/scale/reading",
            {
                "device_id": device_id,
                "weight_grams": weight_grams,
                "stable": stable,
                "raw_adc": raw_adc,
            },
        )

    async def write_tag_result(
        self, device_id: str, spool_id: int, tag_uid: str, success: bool, message: str | None = None
    ) -> dict | None:
        return await self._post(
            WRITE_RESULT_PATH,
            {
                "device_id": device_id,
                "spool_id": spool_id,
                "tag_uid": tag_uid,
                "success": success,
                "message": message,
            },
        )

    async def diagnostic_result(
        self,
        device_id: str,
        diagnostic: str,
        success: bool,
        output: str,
        exit_code: int,
    ) -> dict | None:
        return await self._post(
            f"/diagnostics/{device_id}/result",
            {
                "diagnostic": diagnostic,
                "success": success,
                "output": output,
                "exit_code": exit_code,
            },
        )

    async def system_command_result(
        self,
        device_id: str,
        command: str,
        success: bool,
        message: str | None = None,
    ) -> dict | None:
        return await self._post(
            f"/devices/{device_id}/system/command-result",
            {
                "command": command,
                "success": success,
                "message": message,
            },
        )