adebaumann/Selsyn_inspired_gauge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dev #1

Merged
adebaumann merged 9 commits from dev into main 2026-04-11 17:47:38 +00:00
Conversation 0 Commits 9 Files Changed 27 +228 -174
2 changed files with 228 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 916ddaf423 - Show all commits

19
config.multi.example.json
View File

@@ -17,7 +17,11 @@
"max": 7300,
"step_us": 200,
"entity_name": "Power 1",
"unit": "W"
"unit": "W",
"leds": {
"red_pin": 33,
"green_pin": 32
}
},
{
"name": "Gauge 2",
@@ -27,14 +31,17 @@
"max": 100,
"step_us": 200,
"entity_name": "Temperature",
"unit": "C"
"unit": "C",
"leds": {
"red_pin": 21,
"green_pin": 20
}
}
],
"leds": {
"red_pin": 33,
"green_pin": 32,
"backlight_pin": 23
"backlight": {
"pin": 23,
"num_leds_per_gauge": 3
},
"device": {

383
gaugemqttcontinuous.py
View File

@@ -88,11 +88,9 @@ MICROSTEPS_PER_SECOND = int(_cfg.get("microsteps_per_second", 600))
HEARTBEAT_MS = int(_cfg.get("heartbeat_ms", 10000))
REZERO_INTERVAL_MS = int(_cfg.get("rezero_interval_ms", 3600000))
LED_RED_PIN = int(_cfg.get("leds", {}).get("red_pin", _cfg.get("led_red_pin", 33)))
LED_GREEN_PIN = int(
_cfg.get("leds", {}).get("green_pin", _cfg.get("led_green_pin", 32))
)
LED_BL_PIN = int(_cfg.get("leds", {}).get("backlight_pin", _cfg.get("led_bl_pin", 23)))
backlight_cfg = _cfg.get("backlight", {})
BACKLIGHT_PIN = int(backlight_cfg.get("pin", _cfg.get("led_bl_pin", 23)))
BACKLIGHT_LEDS_PER_GAUGE = int(backlight_cfg.get("num_leds_per_gauge", 3))
device_cfg = _cfg.get("device", {})
DEVICE_NAME = device_cfg.get("name", _cfg.get("device_name", "Selsyn Multi"))
@@ -107,6 +105,7 @@ DEVICE_AREA = device_cfg.get("area", _cfg.get("device_area", "Control Panels"))
gauges = []
if "gauges" in _cfg:
for i, g in enumerate(_cfg["gauges"]):
led_cfg = g.get("leds", {})
gauges.append(
{
"id": i,
@@ -118,6 +117,8 @@ if "gauges" in _cfg:
"step_us": int(g.get("step_us", 200)),
"entity_name": g.get("entity_name", f"Gauge {i + 1}"),
"unit": g.get("unit", ""),
"red_pin": int(led_cfg.get("red_pin", 33)),
"green_pin": int(led_cfg.get("green_pin", 32)),
}
)
else:
@@ -132,10 +133,10 @@ else:
"step_us": int(_cfg.get("gauge_step_us", 200)),
"entity_name": _cfg.get("gauge_entity_name", "Selsyn 1 Power"),
"unit": _cfg.get("gauge_unit", "W"),
"red_pin": int(_cfg.get("led_red_pin", 33)),
"green_pin": int(_cfg.get("led_green_pin", 32)),
}
)
BL_ENTITY = _cfg.get("backlight_entity_name", "Selsyn Backlight")
BL_UNIT = _cfg.get("backlight_unit", "%")
# ---------------------------------------------------------------------------
@@ -172,6 +173,15 @@ def make_gauge_topics(prefix, gauge_id):
"status": f"{prefix}/gauge{gauge_id}/status",
"zero": f"{prefix}/gauge{gauge_id}/zero",
"disc": f"homeassistant/number/{MQTT_CLIENT_ID}_g{gauge_id}/config",
"led_red": f"{prefix}/gauge{gauge_id}/led/red/set",
"led_green": f"{prefix}/gauge{gauge_id}/led/green/set",
"led_bl": f"{prefix}/gauge{gauge_id}/led/backlight/set",
"led_red_state": f"{prefix}/gauge{gauge_id}/led/red/state",
"led_green_state": f"{prefix}/gauge{gauge_id}/led/green/state",
"led_bl_state": f"{prefix}/gauge{gauge_id}/led/backlight/state",
"led_red_disc": f"homeassistant/switch/{MQTT_CLIENT_ID}_g{gauge_id}_red/config",
"led_green_disc": f"homeassistant/switch/{MQTT_CLIENT_ID}_g{gauge_id}_green/config",
"led_bl_disc": f"homeassistant/light/{MQTT_CLIENT_ID}_g{gauge_id}_bl/config",
}
@@ -181,18 +191,40 @@ T_SET = f"{MQTT_PREFIX}/set"
T_STATE = f"{MQTT_PREFIX}/state"
T_STATUS = f"{MQTT_PREFIX}/status"
T_ZERO = f"{MQTT_PREFIX}/zero"
T_LED_RED = f"{MQTT_PREFIX}/led/red/set"
T_LED_GREEN = f"{MQTT_PREFIX}/led/green/set"
T_LED_BL = f"{MQTT_PREFIX}/led/backlight/set"
T_LED_BL_STATE = f"{MQTT_PREFIX}/led/backlight/state"
T_LED_RED_STATE = f"{MQTT_PREFIX}/led/red/state"
T_LED_GREEN_STATE = f"{MQTT_PREFIX}/led/green/state"
gauge_topics = [make_gauge_topics(MQTT_PREFIX, g["id"]) for g in gauges]
T_SET = f"{MQTT_PREFIX}/set"
T_STATE = f"{MQTT_PREFIX}/state"
T_STATUS = f"{MQTT_PREFIX}/status"
T_ZERO = f"{MQTT_PREFIX}/zero"
T_DISC_GAUGE = f"homeassistant/number/{MQTT_CLIENT_ID}/config"
T_DISC_RED = f"homeassistant/switch/{MQTT_CLIENT_ID}_red/config"
T_DISC_GREEN = f"homeassistant/switch/{MQTT_CLIENT_ID}_green/config"
T_DISC_BL = f"homeassistant/light/{MQTT_CLIENT_ID}_bl/config"
def make_gauge_topics(prefix, gauge_id):
return {
"set": f"{prefix}/gauge{gauge_id}/set",
"state": f"{prefix}/gauge{gauge_id}/state",
"status": f"{prefix}/gauge{gauge_id}/status",
"zero": f"{prefix}/gauge{gauge_id}/zero",
"disc": f"homeassistant/number/{MQTT_CLIENT_ID}_g{gauge_id}/config",
"led_red": f"{prefix}/gauge{gauge_id}/led/red/set",
"led_green": f"{prefix}/gauge{gauge_id}/led/green/set",
"led_bl": f"{prefix}/gauge{gauge_id}/led/backlight/set",
"led_red_state": f"{prefix}/gauge{gauge_id}/led/red/state",
"led_green_state": f"{prefix}/gauge{gauge_id}/led/green/state",
"led_bl_state": f"{prefix}/gauge{gauge_id}/led/backlight/state",
"led_red_disc": f"homeassistant/switch/{MQTT_CLIENT_ID}_g{gauge_id}_red/config",
"led_green_disc": f"homeassistant/switch/{MQTT_CLIENT_ID}_g{gauge_id}_green/config",
"led_bl_disc": f"homeassistant/light/{MQTT_CLIENT_ID}_g{gauge_id}_bl/config",
}
_DEVICE = {
"identifiers": [MQTT_CLIENT_ID],
"name": DEVICE_NAME,
@@ -267,42 +299,58 @@ def check_wifi():
# ---------------------------------------------------------------------------
# LEDs
# LEDs (per gauge)
# ---------------------------------------------------------------------------
led_red = Pin(LED_RED_PIN, Pin.OUT, value=0)
led_green = Pin(LED_GREEN_PIN, Pin.OUT, value=0)
led_bl = NeoPixel(Pin(LED_BL_PIN), 3)
num_gauges = len(gauges)
leds_red = []
leds_green = []
leds_bl = []
for g in gauges:
leds_red.append(Pin(g["red_pin"], Pin.OUT, value=0))
leds_green.append(Pin(g["green_pin"], Pin.OUT, value=0))
total_backlight_leds = num_gauges * BACKLIGHT_LEDS_PER_GAUGE
leds_bl = NeoPixel(Pin(BACKLIGHT_PIN), total_backlight_leds)
_backlight_color = (0, 0, 0)
_backlight_brightness = 100 # last *active* brightness — never set to 0
_backlight_brightness = 100
_backlight_on = False
_bl_dirty_since = None
_BL_SAVE_DELAY_MS = 5000
backlight_color = [(0, 0, 0) for _ in range(num_gauges)]
backlight_brightness = [100 for _ in range(num_gauges)]
backlight_on = [False for _ in range(num_gauges)]
def _flush_backlight(client):
payload = {
"state": "ON" if _backlight_on else "OFF",
"color": {
"r": _backlight_color[0],
"g": _backlight_color[1],
"b": _backlight_color[2],
},
"brightness": int(_backlight_brightness * 2.55),
}
client.publish(T_LED_BL, ujson.dumps(payload), retain=True)
info(
f"Backlight state retained: {payload['state']} {_backlight_color} @ {_backlight_brightness}%"
)
for i in range(num_gauges):
gt = gauge_topics[i]
payload = {
"state": "ON" if backlight_on[i] else "OFF",
"color": {
"r": backlight_color[i][0],
"g": backlight_color[i][1],
"b": backlight_color[i][2],
},
"brightness": int(backlight_brightness[i] * 2.55),
}
client.publish(
f"{gt['set'].replace('/set', '/backlight/state')}",
ujson.dumps(payload),
retain=True,
)
info(
f"Gauge {i} backlight: {payload['state']} {backlight_color[i]} @ {backlight_brightness[i]}%"
)
def _backlight_changed(new_color, new_on, new_brightness):
"""Return True if any backlight property differs from current state."""
def _backlight_changed(gauge_idx, new_color, new_on, new_brightness):
return (
new_color != _backlight_color
or new_on != _backlight_on
or (new_on and new_brightness != _backlight_brightness)
new_color != backlight_color[gauge_idx]
or new_on != backlight_on[gauge_idx]
or (new_on and new_brightness != backlight_brightness[gauge_idx])
)
@@ -311,43 +359,44 @@ def _mark_bl_dirty():
_bl_dirty_since = utime.ticks_ms()
def set_backlight_color(r, g, b, brightness=None):
global _backlight_color, _backlight_brightness, _backlight_on
def set_backlight_color(gauge_idx, r, g, b, brightness=None):
global backlight_color, backlight_brightness, backlight_on
if brightness is None:
brightness = _backlight_brightness
brightness = backlight_brightness[gauge_idx]
new_on = brightness > 0
if not _backlight_changed((r, g, b), new_on, brightness):
if not _backlight_changed(gauge_idx, (r, g, b), new_on, brightness):
return
_backlight_color = (r, g, b)
backlight_color[gauge_idx] = (r, g, b)
if brightness > 0:
_backlight_brightness = brightness
_backlight_on = new_on
backlight_brightness[gauge_idx] = brightness
backlight_on[gauge_idx] = new_on
scale = brightness / 100
for i in range(3):
led_bl[i] = (int(g * scale), int(r * scale), int(b * scale))
led_bl.write()
base_idx = gauge_idx * BACKLIGHT_LEDS_PER_GAUGE
for j in range(BACKLIGHT_LEDS_PER_GAUGE):
leds_bl[base_idx + j] = (int(g * scale), int(r * scale), int(b * scale))
leds_bl.write()
_mark_bl_dirty()
def set_backlight_brightness(brightness):
global _backlight_brightness, _backlight_on
def set_backlight_brightness(gauge_idx, brightness):
global backlight_brightness, backlight_on
clamped = max(0, min(100, brightness))
new_on = clamped > 0
if not _backlight_changed(_backlight_color, new_on, clamped):
if not _backlight_changed(gauge_idx, backlight_color[gauge_idx], new_on, clamped):
return
if clamped > 0:
_backlight_brightness = clamped
_backlight_on = new_on
r, g, b = _backlight_color
backlight_brightness[gauge_idx] = clamped
backlight_on[gauge_idx] = new_on
r, g, b = backlight_color[gauge_idx]
scale = clamped / 100
for i in range(3):
led_bl[i] = (int(g * scale), int(r * scale), int(b * scale))
led_bl.write()
base_idx = gauge_idx * BACKLIGHT_LEDS_PER_GAUGE
for j in range(BACKLIGHT_LEDS_PER_GAUGE):
leds_bl[base_idx + j] = (int(g * scale), int(r * scale), int(b * scale))
leds_bl.write()
_mark_bl_dirty()
info(f"LEDs ready red={LED_RED_PIN} green={LED_GREEN_PIN} backlight={LED_BL_PIN}")
# ---------------------------------------------------------------------------
# State
# ---------------------------------------------------------------------------
@@ -407,58 +456,55 @@ def on_message(topic, payload):
info("All gauges zeroed")
return
if topic == T_LED_RED:
state = payload.upper() == "ON"
led_red.value(1 if state else 0)
_publish(T_LED_RED_STATE, "ON" if state else "OFF", retain=True)
info(f"Red LED → {'ON' if state else 'OFF'}")
return
if topic == T_LED_GREEN:
state = payload.upper() == "ON"
led_green.value(1 if state else 0)
_publish(T_LED_GREEN_STATE, "ON" if state else "OFF", retain=True)
info(f"Green LED → {'ON' if state else 'OFF'}")
return
if topic == T_LED_BL:
info(f"Backlight raw payload: '{payload}'")
try:
data = ujson.loads(payload)
info(
f"Backlight parsed: state={data.get('state')} color={data.get('color')} brightness={data.get('brightness')}"
)
if data.get("state", "ON").upper() == "OFF":
set_backlight_brightness(0)
_publish(T_LED_BL_STATE, ujson.dumps({"state": "OFF"}), retain=True)
info("Backlight → OFF")
return
color = data.get("color", {})
r = max(0, min(255, int(color.get("r", _backlight_color[0]))))
g = max(0, min(255, int(color.get("g", _backlight_color[1]))))
b = max(0, min(255, int(color.get("b", _backlight_color[2]))))
# HA sends brightness as 0-255; convert to 0-100
raw_br = data.get("brightness", None)
if raw_br is not None:
brightness = max(0, min(100, round(int(raw_br) / 2.55)))
elif _backlight_brightness > 0:
brightness = _backlight_brightness
else:
brightness = 100
except Exception as e:
warn(f"Invalid backlight payload: '{payload}' ({e})")
for i, gt in enumerate(gauge_topics):
if topic == gt["led_red"]:
state = payload.upper() == "ON"
leds_red[i].value(1 if state else 0)
_publish(gt["led_red_state"], "ON" if state else "OFF", retain=True)
info(f"Gauge {i} red LED → {'ON' if state else 'OFF'}")
return
if topic == gt["led_green"]:
state = payload.upper() == "ON"
leds_green[i].value(1 if state else 0)
_publish(gt["led_green_state"], "ON" if state else "OFF", retain=True)
info(f"Gauge {i} green LED → {'ON' if state else 'OFF'}")
return
if topic == gt["led_bl"]:
try:
data = ujson.loads(payload)
if data.get("state", "ON").upper() == "OFF":
set_backlight_brightness(i, 0)
_publish(
gt["led_bl_state"], ujson.dumps({"state": "OFF"}), retain=True
)
info(f"Gauge {i} backlight → OFF")
return
color = data.get("color", {})
r = max(0, min(255, int(color.get("r", backlight_color[i][0]))))
g = max(0, min(255, int(color.get("g", backlight_color[i][1]))))
b = max(0, min(255, int(color.get("b", backlight_color[i][2]))))
raw_br = data.get("brightness", None)
if raw_br is not None:
brightness = max(0, min(100, round(int(raw_br) / 2.55)))
elif backlight_brightness[i] > 0:
brightness = backlight_brightness[i]
else:
brightness = 100
except Exception as e:
warn(f"Invalid backlight payload for gauge {i}: '{payload}' ({e})")
return
set_backlight_color(i, r, g, b, brightness)
state = {
"state": "ON",
"color_mode": "rgb",
"brightness": int(brightness * 2.55),
"color": {"r": r, "g": g, "b": b},
}
_publish(gt["led_bl_state"], ujson.dumps(state), retain=True)
info(f"Gauge {i} backlight → #{r:02x}{g:02x}{b:02x} @ {brightness}%")
return
set_backlight_color(r, g, b, brightness)
color_hex = f"#{r:02x}{g:02x}{b:02x}"
state = {
"state": "ON",
"color_mode": "rgb",
"brightness": int(brightness * 2.55),
"color": {"r": r, "g": g, "b": b},
}
_publish(T_LED_BL_STATE, ujson.dumps(state), retain=True)
info(f"Backlight → {color_hex} @ {brightness}%")
return
# ---------------------------------------------------------------------------
@@ -483,12 +529,12 @@ def connect_mqtt():
client_ref = client
client.subscribe(T_SET)
client.subscribe(T_ZERO)
client.subscribe(T_LED_RED)
client.subscribe(T_LED_GREEN)
client.subscribe(T_LED_BL)
for gt in gauge_topics:
client.subscribe(gt["set"])
client.subscribe(gt["zero"])
client.subscribe(gt["led_red"])
client.subscribe(gt["led_green"])
client.subscribe(gt["led_bl"])
_mqtt_connected = True
info(f"MQTT connected client_id={MQTT_CLIENT_ID}")
return client
@@ -533,12 +579,12 @@ def check_mqtt():
client_ref.connect()
client_ref.subscribe(T_SET)
client_ref.subscribe(T_ZERO)
client_ref.subscribe(T_LED_RED)
client_ref.subscribe(T_LED_GREEN)
client_ref.subscribe(T_LED_BL)
for gt in gauge_topics:
client_ref.subscribe(gt["set"])
client_ref.subscribe(gt["zero"])
client_ref.subscribe(gt["led_red"])
client_ref.subscribe(gt["led_green"])
client_ref.subscribe(gt["led_bl"])
_mqtt_connected = True
info("MQTT reconnected!")
publish_discovery(client_ref)
@@ -558,6 +604,7 @@ def publish_discovery(client):
for i, g in enumerate(gauges):
gt = gauge_topics[i]
client.publish(
gt["disc"],
ujson.dumps(
@@ -579,62 +626,62 @@ def publish_discovery(client):
)
info(f"Discovery: gauge {i} ({g['name']})")
client.publish(
T_DISC_RED,
ujson.dumps(
{
"name": "Red LED",
"uniq_id": f"{MQTT_CLIENT_ID}_led_red",
"cmd_t": T_LED_RED,
"stat_t": T_LED_RED_STATE,
"pl_on": "ON",
"pl_off": "OFF",
"icon": "mdi:led-on",
"dev": _dev_ref,
"ret": True,
}
),
retain=True,
)
info("Discovery: red LED")
client.publish(
gt["led_red_disc"],
ujson.dumps(
{
"name": f"{g['name']} Red LED",
"uniq_id": f"{MQTT_CLIENT_ID}_g{i}_red",
"cmd_t": gt["led_red"],
"stat_t": gt["led_red_state"],
"pl_on": "ON",
"pl_off": "OFF",
"icon": "mdi:led-on",
"dev": _dev_ref,
"ret": True,
}
),
retain=True,
)
info(f"Discovery: gauge {i} red LED")
client.publish(
T_DISC_GREEN,
ujson.dumps(
{
"name": "Green LED",
"uniq_id": f"{MQTT_CLIENT_ID}_led_green",
"cmd_t": T_LED_GREEN,
"stat_t": T_LED_GREEN_STATE,
"pl_on": "ON",
"pl_off": "OFF",
"icon": "mdi:led-on",
"dev": _dev_ref,
"ret": True,
}
),
retain=True,
)
info("Discovery: green LED")
client.publish(
gt["led_green_disc"],
ujson.dumps(
{
"name": f"{g['name']} Green LED",
"uniq_id": f"{MQTT_CLIENT_ID}_g{i}_green",
"cmd_t": gt["led_green"],
"stat_t": gt["led_green_state"],
"pl_on": "ON",
"pl_off": "OFF",
"icon": "mdi:led-on",
"dev": _dev_ref,
"ret": True,
}
),
retain=True,
)
info(f"Discovery: gauge {i} green LED")
client.publish(
T_DISC_BL,
ujson.dumps(
{
"name": BL_ENTITY,
"uniq_id": f"{MQTT_CLIENT_ID}_led_bl",
"cmd_t": T_LED_BL,
"stat_t": T_LED_BL_STATE,
"schema": "json",
"supported_color_modes": ["rgb"],
"icon": "mdi:led-strip",
"dev": _dev_ref,
"ret": True,
}
),
retain=True,
)
info("Discovery: backlight")
client.publish(
gt["led_bl_disc"],
ujson.dumps(
{
"name": f"{g['name']} Backlight",
"uniq_id": f"{MQTT_CLIENT_ID}_g{i}_bl",
"cmd_t": gt["led_bl"],
"stat_t": gt["led_bl_state"],
"schema": "json",
"supported_color_modes": ["rgb"],
"icon": "mdi:led-strip",
"dev": _dev_ref,
"ret": True,
}
),
retain=True,
)
info(f"Discovery: gauge {i} backlight")
def publish_state(client):