""" 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 _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 client = connect_mqtt() publish_discovery(client) publish_state(client) 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.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()