From 916ddaf423d129df3534da12560a181860551991 Mon Sep 17 00:00:00 2001 From: "Adrian A. Baumann" Date: Fri, 10 Apr 2026 23:50:06 +0200 Subject: [PATCH] LEDs and backlight split up as well --- config.multi.example.json | 19 +- gaugemqttcontinuous.py | 383 +++++++++++++++++++++----------------- 2 files changed, 228 insertions(+), 174 deletions(-) diff --git a/config.multi.example.json b/config.multi.example.json index 93191c4..8cd14ff 100644 --- a/config.multi.example.json +++ b/config.multi.example.json @@ -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": { diff --git a/gaugemqttcontinuous.py b/gaugemqttcontinuous.py index 6cc2c3d..e16db2e 100644 --- a/gaugemqttcontinuous.py +++ b/gaugemqttcontinuous.py @@ -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):