""" kxgauge_client.py - thin synchronous HTTP client for KXGauge (ESP32 round-face display, https://gitea.it-drui.de/viewit/kxgauge). KXGauge exposes a simple GET-only HTTP API (no push, no websocket) - the bridge has to actively poke it whenever printer state/temperature changes. Designed to be called from daemon threads (MQTT reader thread callbacks), mirrors spoolman_client.py's shape: plain `requests`, no event-loop dependency. ──────────────────────────────────────────────────────────────────────────── Copyright (C) 2026 viewit (KX-Bridge contributors) Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md. """ import logging log = logging.getLogger("kobrax.kxgauge") # Tolerance in °C before a new heat value is worth another GET - avoids # spamming the device on every MQTT tick when the temperature is basically # holding steady. _HEAT_TOLERANCE = 2.0 class KXGaugeClient: """Thin synchronous HTTP client for a KXGauge display. All calls swallow their own exceptions and just log a warning - a disconnected/misconfigured display must never interrupt MQTT callback processing or the poll loop. """ def __init__(self, base_url: str, heat_peak: float = 250.0): self.base_url = base_url.rstrip("/") self.heat_peak = heat_peak self._last_emotion: str | None = None self._last_heat_celsius: float | None = None self._peak_sent = False def _get(self, path: str) -> bool: try: import requests r = requests.get(f"{self.base_url}{path}", timeout=3) r.raise_for_status() return True except Exception as e: log.warning(f"KXGauge request failed ({path}): {e}") return False def health_check(self) -> bool: try: import requests r = requests.get(f"{self.base_url}/status", timeout=3) r.raise_for_status() return True except Exception: return False def set_emotion(self, name: str) -> None: name = (name or "").lower().strip() if not name or name == self._last_emotion: return if self._get(f"/emotion/{name}"): self._last_emotion = name def set_heat_celsius(self, value: float) -> None: if not self._peak_sent: if self._get(f"/heat/peak/celsius/{self.heat_peak}"): self._peak_sent = True if (self._last_heat_celsius is not None and abs(value - self._last_heat_celsius) < _HEAT_TOLERANCE): return if self._get(f"/heat/celsius/{value:.1f}"): self._last_heat_celsius = value