viewit
6ab5c31035
All checks were successful
Testing Build / build (push) Successful in 8m44s
Adds an optional per-printer integration with KXGauge (https://gitea.it-drui.de/viewit/kxgauge), a small ESP32 round-face display that shows printer status as an emotion and hotend temperature as a color ring. KXGauge only exposes a GET-only HTTP API with no push/websocket, so the bridge actively pushes to it from the existing MQTT callbacks (_on_temp for the heat ring, _on_print + offline transitions for the emotion) whenever state actually changes, with a built-in dedupe so it doesn't spam the device every poll tick. - kxgauge_client.py: thin synchronous HTTP client (mirrors spoolman_client.py's shape - called from the MQTT reader thread). - New [kxgauge]/[kxgauge_mapping] config.ini sections; the mapping (kobra_state -> KXGauge emotion) is user-editable with a sane default and falls back per-key if only partially configured. - Settings UI: new card under Integrations with enable/URL/target-temp fields, a per-state emotion mapping list, and a connection-test button (/api/kxgauge/test). - Multi-printer aware: kxgauge_url/enabled/heat_peak merge per [printer_N] like the existing power-switch settings. Also fixes a real bug found while testing this: _find_config_path() resolves to the live project config/config.ini, not a sandboxed path, so any test hitting /api/settings POST without stubbing it out will silently overwrite the real printer config. test_settings.py already guards against this - test_kxgauge.py now does too.
75 lines
2.8 KiB
Python
75 lines
2.8 KiB
Python
"""
|
|
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
|