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

Status LEDs on control rod number plate added

Browse Source
This commit is contained in:
Adrian A. Baumann
2026-04-11 21:40:08 +02:00
parent bed3bf2b4a
commit fc12d92d8d
4 changed files with 118 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
gaugemqttcontinuous.py
View File

@@ -98,6 +98,7 @@ REZERO_INTERVAL_MS = int(_cfg.get("rezero_interval_ms", 3600000))
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))
STATUS_LEDS_PER_GAUGE = int(backlight_cfg.get("num_status_leds_per_gauge", 2))
device_cfg = _cfg.get("device", {})
DEVICE_NAME = device_cfg.get("name", _cfg.get("device_name", "Selsyn Multi"))
@@ -126,6 +127,8 @@ if "gauges" in _cfg:
"unit": g.get("unit", ""),
"red_pin": int(led_cfg.get("red_pin", 33)),
"green_pin": int(led_cfg.get("green_pin", 32)),
"ws2812_red": tuple(led_cfg.get("ws2812_red", [255, 0, 0])),
"ws2812_green": tuple(led_cfg.get("ws2812_green", [0, 255, 0])),
}
)
else:
@@ -142,6 +145,8 @@ else:
"unit": _cfg.get("gauge_unit", "W"),
"red_pin": int(_cfg.get("led_red_pin", 33)),
"green_pin": int(_cfg.get("led_green_pin", 32)),
"ws2812_red": tuple(_cfg.get("ws2812_red", [255, 0, 0])),
"ws2812_green": tuple(_cfg.get("ws2812_green", [0, 255, 0])),
}
)
BL_UNIT = _cfg.get("backlight_unit", "%")
@@ -229,6 +234,12 @@ def make_gauge_topics(prefix, gauge_id):
"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",
"status_red": f"{prefix}/gauge{gauge_id}/status_led/red/set",
"status_green": f"{prefix}/gauge{gauge_id}/status_led/green/set",
"status_red_state": f"{prefix}/gauge{gauge_id}/status_led/red/state",
"status_green_state": f"{prefix}/gauge{gauge_id}/status_led/green/state",
"status_red_disc": f"homeassistant/switch/{MQTT_CLIENT_ID}_g{gauge_id}_status_red/config",
"status_green_disc": f"homeassistant/switch/{MQTT_CLIENT_ID}_g{gauge_id}_status_green/config",
}
@@ -317,12 +328,16 @@ 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
total_backlight_leds = num_gauges * (BACKLIGHT_LEDS_PER_GAUGE + STATUS_LEDS_PER_GAUGE)
leds_bl = NeoPixel(Pin(BACKLIGHT_PIN), total_backlight_leds)
_backlight_color = (0, 0, 0)
_backlight_brightness = 100
_backlight_on = False
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)]
status_led_red = [False for _ in range(num_gauges)]
status_led_green = [False for _ in range(num_gauges)]
_bl_dirty_since = None
_BL_SAVE_DELAY_MS = 5000
@@ -379,9 +394,11 @@ def set_backlight_color(gauge_idx, r, g, b, brightness=None):
backlight_on[gauge_idx] = new_on
scale = brightness / 100
base_idx = gauge_idx * BACKLIGHT_LEDS_PER_GAUGE
leds_per_gauge = BACKLIGHT_LEDS_PER_GAUGE + STATUS_LEDS_PER_GAUGE
base_idx = gauge_idx * 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))
_update_status_leds(gauge_idx)
leds_bl.write()
_mark_bl_dirty()
@@ -397,13 +414,44 @@ def set_backlight_brightness(gauge_idx, brightness):
backlight_on[gauge_idx] = new_on
r, g, b = backlight_color[gauge_idx]
scale = clamped / 100
base_idx = gauge_idx * BACKLIGHT_LEDS_PER_GAUGE
leds_per_gauge = BACKLIGHT_LEDS_PER_GAUGE + STATUS_LEDS_PER_GAUGE
base_idx = gauge_idx * 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))
_update_status_leds(gauge_idx)
leds_bl.write()
_mark_bl_dirty()
def _update_status_leds(gauge_idx):
leds_per_gauge = BACKLIGHT_LEDS_PER_GAUGE + STATUS_LEDS_PER_GAUGE
base_idx = gauge_idx * leds_per_gauge + BACKLIGHT_LEDS_PER_GAUGE
g_cfg = gauges[gauge_idx]
red_color = g_cfg["ws2812_red"]
green_color = g_cfg["ws2812_green"]
if status_led_red[gauge_idx]:
leds_bl[base_idx] = (red_color[1], red_color[0], red_color[2])
else:
leds_bl[base_idx] = (0, 0, 0)
if status_led_green[gauge_idx]:
leds_bl[base_idx + 1] = (green_color[1], green_color[0], green_color[2])
else:
leds_bl[base_idx + 1] = (0, 0, 0)
def set_status_led(gauge_idx, led_type, state):
global status_led_red, status_led_green
if led_type == "red":
status_led_red[gauge_idx] = state
elif led_type == "green":
status_led_green[gauge_idx] = state
_update_status_leds(gauge_idx)
leds_bl.write()
# ---------------------------------------------------------------------------
# State
# ---------------------------------------------------------------------------
@@ -513,6 +561,20 @@ def on_message(topic, payload):
info(f"Gauge {i} backlight → #{r:02x}{g:02x}{b:02x} @ {brightness}%")
return
if topic == gt["status_red"]:
state = payload.upper() == "ON"
set_status_led(i, "red", state)
_publish(gt["status_red_state"], "ON" if state else "OFF", retain=True)
info(f"Gauge {i} status red → {'ON' if state else 'OFF'}")
return
if topic == gt["status_green"]:
state = payload.upper() == "ON"
set_status_led(i, "green", state)
_publish(gt["status_green_state"], "ON" if state else "OFF", retain=True)
info(f"Gauge {i} status green → {'ON' if state else 'OFF'}")
return
# ---------------------------------------------------------------------------
# MQTT connect + discovery
@@ -542,6 +604,8 @@ def connect_mqtt():
client.subscribe(gt["led_red"])
client.subscribe(gt["led_green"])
client.subscribe(gt["led_bl"])
client.subscribe(gt["status_red"])
client.subscribe(gt["status_green"])
_mqtt_connected = True
info(f"MQTT connected client_id={MQTT_CLIENT_ID}")
return client
@@ -592,6 +656,8 @@ def check_mqtt():
client_ref.subscribe(gt["led_red"])
client_ref.subscribe(gt["led_green"])
client_ref.subscribe(gt["led_bl"])
client_ref.subscribe(gt["status_red"])
client_ref.subscribe(gt["status_green"])
_mqtt_connected = True
info("MQTT reconnected!")
publish_discovery(client_ref)
@@ -696,6 +762,44 @@ def publish_discovery(client):
)
info(f"Discovery: gauge {i} backlight")
client.publish(
gt["status_red_disc"],
ujson.dumps(
{
"name": f"{g['name']} Status Red",
"uniq_id": f"{MQTT_CLIENT_ID}_g{i}_status_red",
"cmd_t": gt["status_red"],
"stat_t": gt["status_red_state"],
"pl_on": "ON",
"pl_off": "OFF",
"icon": "mdi:led-on",
"dev": _dev_ref,
"ret": True,
}
),
retain=True,
)
info(f"Discovery: gauge {i} status red")
client.publish(
gt["status_green_disc"],
ujson.dumps(
{
"name": f"{g['name']} Status Green",
"uniq_id": f"{MQTT_CLIENT_ID}_g{i}_status_green",
"cmd_t": gt["status_green"],
"stat_t": gt["status_green_state"],
"pl_on": "ON",
"pl_off": "OFF",
"icon": "mdi:led-on",
"dev": _dev_ref,
"ret": True,
}
),
retain=True,
)
info(f"Discovery: gauge {i} status green")
def publish_state(client):
for i, g in enumerate(gauge_objects):