adebaumann/Selsyn_inspired_gauge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8adec3a86e951241282928930e5812e0ed689b2c
Selsyn_inspired_gauge/gaugemqttcontinuous.py

686 lines
21 KiB
Python
Raw Blame History

"""
gaugemqtt.py — MQTT-based gauge controller for ESP32 / MicroPython
Deploy these files to the ESP32:
gauge.py — stepper driver
gaugemqtt.py — this file
umqtt/simple.py — MicroPython built-in
umqtt/robust.py — https://raw.githubusercontent.com/micropython/micropython-lib/master/micropython/umqtt.robust/umqtt/robust.py
config.json — configuration (see below)
MQTT topics (all prefixed with mqtt_prefix from config.json):
.../set ← HA publishes target value here
.../state → ESP32 publishes current value
.../status → ESP32 publishes online/offline
.../zero ← publish anything to trigger zero
.../led/red/set ← ON/OFF
.../led/green/set ← ON/OFF
.../led/backlight/set ← 0-100
Serial log format: [HH:MM:SS.mmm] LEVEL message
"""
import network
import utime
import ujson
from umqtt.robust import MQTTClient
from machine import Pin
from neopixel import NeoPixel
from gauge_vid6008 import Gauge
# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
def _ts():
ms = utime.ticks_ms()
return f"{(ms // 3600000) % 24:02d}:{(ms // 60000) % 60:02d}:{(ms // 1000) % 60:02d}.{ms % 1000:03d}"
def log(level, msg):
print(f"[{_ts()}] {level:5s} {msg}")
def info(msg):
log("INFO", msg)
def warn(msg):
log("WARN", msg)
def log_err(msg):
log("ERROR", msg)
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
def _load_config():
try:
with open("/config.json") as f:
cfg = ujson.load(f)
info("Config loaded from /config.json")
return cfg
except OSError:
log_err("config.json not found — cannot continue")
raise
except Exception as e:
log_err(f"config.json parse error: {e} — cannot continue")
raise
_cfg = _load_config()
WIFI_SSID = _cfg["wifi_ssid"]
WIFI_PASSWORD = _cfg["wifi_password"]
MQTT_BROKER = _cfg["mqtt_broker"]
MQTT_PORT = int(_cfg.get("mqtt_port", 1883))
MQTT_USER = _cfg["mqtt_user"]
MQTT_PASSWORD = _cfg["mqtt_password"]
MQTT_CLIENT_ID = _cfg["mqtt_client_id"]
MQTT_PREFIX = _cfg["mqtt_prefix"]
GAUGE_PINS = tuple(_cfg.get("gauge_pins", [12, 13]))
GAUGE_MODE = _cfg.get("gauge_mode", "stepdir")
GAUGE_MIN = float(_cfg.get("gauge_min", 0))
GAUGE_MAX = float(_cfg.get("gauge_max", 7300))
GAUGE_STEP_US = int(_cfg.get("gauge_step_us", 200))
MICROSTEPS_PER_SECOND = 600 # microsteps per second (adjustable)
HEARTBEAT_MS = int(_cfg.get("heartbeat_ms", 10000))
REZERO_INTERVAL_MS = int(_cfg.get("rezero_interval_ms", 3600000))
LED_RED_PIN = int(_cfg.get("led_red_pin", 33))
LED_GREEN_PIN = int(_cfg.get("led_green_pin", 32))
LED_BL_PIN = int(_cfg.get("led_bl_pin", 23))
DEVICE_NAME = _cfg.get("device_name", "Selsyn 1")
DEVICE_MODEL = _cfg.get("device_model", "Chernobyl Selsyn-inspired gauge")
DEVICE_MFR = _cfg.get("device_manufacturer", "AdeBaumann")
DEVICE_AREA = _cfg.get("device_area", "Control Panels")
GAUGE_ENTITY = _cfg.get("gauge_entity_name", "Selsyn 1 Power")
GAUGE_UNIT = _cfg.get("gauge_unit", "W")
RED_ENTITY = _cfg.get("red_led_entity_name", "Selsyn 1 Red LED")
GREEN_ENTITY = _cfg.get("green_led_entity_name", "Selsyn 1 Green LED")
BL_ENTITY = _cfg.get("backlight_entity_name", "Selsyn 1 Backlight")
BL_UNIT = _cfg.get("backlight_unit", "%")
# ---------------------------------------------------------------------------
# Topics
# ---------------------------------------------------------------------------
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"
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"
_DEVICE = {
"identifiers": [MQTT_CLIENT_ID],
"name": DEVICE_NAME,
"model": DEVICE_MODEL,
"manufacturer": DEVICE_MFR,
"suggested_area": DEVICE_AREA,
}
# ---------------------------------------------------------------------------
# WiFi
# ---------------------------------------------------------------------------
_wifi_reconnect_delay_s = 5
_wifi_check_interval_ms = 5000
_last_wifi_check = 0
_wifi_sta = None
def connect_wifi(ssid, password, timeout_s=15):
global _wifi_sta
_wifi_sta = network.WLAN(network.STA_IF)
_wifi_sta.active(True)
if _wifi_sta.isconnected():
ip, mask, gw, dns = _wifi_sta.ifconfig()
info("WiFi already connected")
info(f" IP:{ip} mask:{mask} gw:{gw} dns:{dns}")
return ip
info(f"WiFi connecting to '{ssid}' ...")
_wifi_sta.connect(ssid, password)
deadline = utime.time() + timeout_s
while not _wifi_sta.isconnected():
if utime.time() > deadline:
log_err(f"WiFi connect timeout after {timeout_s}s")
raise OSError("WiFi connect timeout")
utime.sleep_ms(200)
ip, mask, gw, dns = _wifi_sta.ifconfig()
mac = ":".join(f"{b:02x}" for b in _wifi_sta.config("mac"))
info("WiFi connected!")
info(f" SSID : {ssid}")
info(f" MAC : {mac}")
info(f" IP : {ip} mask:{mask} gw:{gw} dns:{dns}")
return ip
def check_wifi():
global _wifi_sta, _last_wifi_check, _wifi_reconnect_delay_s
now = utime.ticks_ms()
if utime.ticks_diff(now, _last_wifi_check) < _wifi_check_interval_ms:
return
_last_wifi_check = now
if _wifi_sta is None:
_wifi_sta = network.WLAN(network.STA_IF)
if _wifi_sta.isconnected():
return
log_err("WiFi lost connection — attempting reconnect...")
try:
_wifi_sta.active(True)
_wifi_sta.connect(WIFI_SSID, WIFI_PASSWORD)
deadline = utime.time() + 15
while not _wifi_sta.isconnected():
if utime.time() > deadline:
log_err("WiFi reconnect timeout")
return
utime.sleep_ms(200)
ip, mask, gw, dns = _wifi_sta.ifconfig()
info(f"WiFi reconnected! IP:{ip}")
except Exception as e:
log_err(f"WiFi reconnect failed: {e}")
# ---------------------------------------------------------------------------
# Gauge
# ---------------------------------------------------------------------------
info("Initialising gauge ...")
gauge = Gauge(
pins=GAUGE_PINS,
mode=GAUGE_MODE,
min_val=GAUGE_MIN,
max_val=GAUGE_MAX,
step_us=GAUGE_STEP_US,
)
info(
f"Gauge ready pins={GAUGE_PINS} mode={GAUGE_MODE} range=[{GAUGE_MIN}, {GAUGE_MAX}] step_us={GAUGE_STEP_US}"
)
# ---------------------------------------------------------------------------
# LEDs
# ---------------------------------------------------------------------------
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)
_backlight_color = (0, 0, 0)
_backlight_brightness = 100 # last *active* brightness — never set to 0
_backlight_on = False
_bl_dirty_since = None
_BL_SAVE_DELAY_MS = 5000
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}%"
)
def _backlight_changed(new_color, new_on, new_brightness):
"""Return True if any backlight property differs from current state."""
return (
new_color != _backlight_color
or new_on != _backlight_on
or (new_on and new_brightness != _backlight_brightness)
)
def _mark_bl_dirty():
global _bl_dirty_since
_bl_dirty_since = utime.ticks_ms()
def set_backlight_color(r, g, b, brightness=None):
global _backlight_color, _backlight_brightness, _backlight_on
if brightness is None:
brightness = _backlight_brightness
new_on = brightness > 0
if not _backlight_changed((r, g, b), new_on, brightness):
return
_backlight_color = (r, g, b)
if brightness > 0:
_backlight_brightness = brightness
_backlight_on = 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()
_mark_bl_dirty()
def set_backlight_brightness(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):
return
if clamped > 0:
_backlight_brightness = clamped
_backlight_on = new_on
r, g, b = _backlight_color
scale = clamped / 100
for i in range(3):
led_bl[i] = (int(g * scale), int(r * scale), int(b * scale))
led_bl.write()
_mark_bl_dirty()
info(f"LEDs ready red={LED_RED_PIN} green={LED_GREEN_PIN} backlight={LED_BL_PIN}")
# ---------------------------------------------------------------------------
# State
# ---------------------------------------------------------------------------
_target_value = GAUGE_MIN
_last_rezero_ms = None # set to ticks_ms() in main()
client_ref = None
_mqtt_connected = False
def _publish(topic, payload, retain=False):
"""Safely publish MQTT message, returning True on success."""
if client_ref is None:
return False
try:
client_ref.publish(topic, payload, retain=retain)
return True
except Exception as e:
log_err(f"MQTT publish failed: {e}")
return False
# ---------------------------------------------------------------------------
# MQTT callbacks
# ---------------------------------------------------------------------------
def on_message(topic, payload):
if client_ref is None:
return
topic = topic.decode()
payload = payload.decode().strip()
info(f"MQTT rx {topic} {payload}")
if topic == T_ZERO:
global _last_rezero_ms
info("Zero command received")
gauge.zero()
_last_rezero_ms = utime.ticks_ms()
info("Zero complete")
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})")
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
if topic == T_SET:
global _target_value
try:
_target_value = max(GAUGE_MIN, min(GAUGE_MAX, float(payload)))
info(f"New target → {_target_value:.1f}")
except ValueError:
warn(f"Invalid set value: '{payload}'")
# ---------------------------------------------------------------------------
# MQTT connect + discovery
# ---------------------------------------------------------------------------
def connect_mqtt():
global client_ref, _mqtt_connected
info(f"Connecting to MQTT broker {MQTT_BROKER}:{MQTT_PORT} ...")
client = MQTTClient(
client_id=MQTT_CLIENT_ID,
server=MQTT_BROKER,
port=MQTT_PORT,
user=MQTT_USER,
password=MQTT_PASSWORD,
keepalive=60,
)
client.set_last_will(T_STATUS, b"offline", retain=True, qos=0)
client.set_callback(on_message)
client.connect()
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)
_mqtt_connected = True
info(f"MQTT connected client_id={MQTT_CLIENT_ID}")
return client
def check_mqtt():
global client_ref, _mqtt_connected
if client_ref is None:
return False
try:
client_ref.ping()
_mqtt_connected = True
return True
except Exception as e:
log_err(f"MQTT connection lost: {e}")
_mqtt_connected = False
# Try to reconnect
log_err("Attempting MQTT reconnection...")
for attempt in range(3):
try:
client_ref = MQTTClient(
client_id=MQTT_CLIENT_ID,
server=MQTT_BROKER,
port=MQTT_PORT,
user=MQTT_USER,
password=MQTT_PASSWORD,
keepalive=60,
)
client_ref.set_last_will(T_STATUS, b"offline", retain=True, qos=0)
client_ref.set_callback(on_message)
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)
_mqtt_connected = True
info("MQTT reconnected!")
publish_discovery(client_ref)
publish_state(client_ref)
return True
except Exception as e2:
log_err(f"MQTT reconnect attempt {attempt + 1} failed: {e2}")
utime.sleep_ms(2000)
log_err("MQTT reconnection failed after 3 attempts")
return False
def publish_discovery(client):
"""Publish all HA MQTT discovery payloads using short-form keys to stay under 512 bytes."""
# Full device block only on first payload; subsequent use identifiers-only ref
_dev_ref = {"identifiers": [MQTT_CLIENT_ID]}
client.publish(
T_DISC_GAUGE,
ujson.dumps(
{
"name": GAUGE_ENTITY,
"unique_id": MQTT_CLIENT_ID,
"cmd_t": T_SET,
"stat_t": T_STATE,
"avty_t": T_STATUS,
"min": GAUGE_MIN,
"max": GAUGE_MAX,
"step": 1,
"unit_of_meas": GAUGE_UNIT,
"icon": "mdi:gauge",
"dev": _DEVICE,
}
),
retain=True,
)
info("Discovery: gauge")
client.publish(
T_DISC_RED,
ujson.dumps(
{
"name": RED_ENTITY,
"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(
T_DISC_GREEN,
ujson.dumps(
{
"name": GREEN_ENTITY,
"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(
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")
def publish_state(client):
val = gauge.get()
client.publish(T_STATE, str(round(val, 1)), retain=True)
client.publish(T_STATUS, "online", retain=True)
info(f"State published value={val:.1f} step={gauge._current_step}")
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
info("=" * 48)
info("Gauge MQTT controller starting")
info("=" * 48)
connect_wifi(WIFI_SSID, WIFI_PASSWORD)
info("Zeroing gauge on startup ...")
gauge.zero()
info("Zero complete")
# umqtt.robust handles reconnection automatically — just connect once
connect_mqtt()
publish_discovery(client_ref)
publish_state(client_ref)
info("Entering main loop")
info("-" * 48)
try:
import ota
ota.mark_ok()
info("OTA OK flag set")
except ImportError:
pass
global _last_rezero_ms, _bl_dirty_since
last_heartbeat = utime.ticks_ms()
_last_rezero_ms = utime.ticks_ms()
target_step = gauge._val_to_step(_target_value)
# Period at which to publish state updates during movement
MOVE_STATE_INTERVAL_MS = 500
last_move_state = utime.ticks_ms()
while True:
try:
check_wifi()
if not check_mqtt():
utime.sleep_ms(1000)
continue
client_ref.check_msg()
now = utime.ticks_ms()
# Continuously move towards target at constant speed
current_target = gauge._val_to_step(_target_value)
moved = False
if current_target != gauge._current_step:
direction = 1 if current_target > gauge._current_step else -1
gauge.step(direction)
moved = True
# Publish state during movement at intervals
if (
moved
and utime.ticks_diff(now, last_move_state) >= MOVE_STATE_INTERVAL_MS
):
publish_state(client_ref)
last_move_state = now
# Sleep to achieve constant speed
if moved or (
current_target == gauge._current_step
and gauge._current_step != gauge._val_to_step(_target_value)
):
delay_us = 1_000_000 // MICROSTEPS_PER_SECOND
utime.sleep_us(delay_us)
# Periodic auto-rezero (disabled when interval is 0)
if (
REZERO_INTERVAL_MS > 0
and utime.ticks_diff(now, _last_rezero_ms) >= REZERO_INTERVAL_MS
):
info("Auto-rezero triggered")
saved = _target_value
gauge.zero()
if saved > GAUGE_MIN:
gauge.set(saved)
publish_state(client_ref)
_last_rezero_ms = now
info(f"Auto-rezero complete, restored to {saved:.1f}")
# Retain backlight state via MQTT after settling
if (
_bl_dirty_since is not None
and utime.ticks_diff(now, _bl_dirty_since) >= _BL_SAVE_DELAY_MS
):
_flush_backlight(client_ref)
_bl_dirty_since = None
# Heartbeat
if utime.ticks_diff(now, last_heartbeat) >= HEARTBEAT_MS:
publish_state(client_ref)
info(
f"step={gauge._current_step} phase={gauge._phase} target_step={gauge._val_to_step(_target_value)} expected_phase={gauge._current_step % 4}"
)
last_heartbeat = now
except Exception as e:
log_err(f"Main loop error: {e} — continuing")
utime.sleep_ms(100)
main()
Reference in New Issue View Git Blame Copy Permalink