#!/usr/bin/python3 -u # Publish Lithionics Li3 BMS battery telemetry to Home Assistant via MQTT # discovery. Protocol reverse-engineered from the com.lithionics.bms Android # app (BLEMaster / MainBmsCsParameters): classic HM-10 BLE-UART (service # ffe0, characteristic ffe1, notify+write, no pairing). On connect we send # "$traceon" then "$info"; the device then streams CRLF-terminated CSV # telemetry lines forever. # # BLE connect failures are retried internally (MQTT session stays up across # them) rather than exiting, because bluetoothd on this Pi's onboard adapter # occasionally wedges its discovery state after a run of failed connection # attempts (Discovering stays "yes" forever, and every subsequent connect # fails with le-connection-abort-by-local) — observed happening from our own # repeated scan/connect cycling, not anything external. We restart # bluetooth.service ourselves once scan misses run past STUCK_DISCOVERY_THRESHOLD # consecutive attempts, rate-limited (BLUETOOTH_RESTART_COOLDOWN_S) so we don't do # it so often it disrupts the other BLE gear on this hub (motion sensors, IR # remote, etc. also served by bluetoothd here). # # The battery itself sits at the edge of the onboard adapter's range (around # -71 to -73 dBm from its fixed position, confirmed 2026-08-24 against a phone # that connects to the same device instantly from up close) — an ordinary scan # miss from weak RSSI looks identical to a wedged discovery state, so # STUCK_DISCOVERY_THRESHOLD is set well above 1 to avoid restarting bluetoothd # (which doesn't fix an RF problem) on every routine miss. # # Only exits (letting systemd Restart=always give us a fresh process) on # MQTT-level failure, which shouldn't happen in normal operation. import asyncio import json import subprocess import sys import time import paho.mqtt.client as mqtt from bleak import BleakClient, BleakScanner BLE_RETRY_DELAY_S = 8 STUCK_DISCOVERY_THRESHOLD = 10 # consecutive scan failures before we intervene BLUETOOTH_RESTART_COOLDOWN_S = 300 # don't restart bluetooth.service more than this often MQTT_STARTUP_TIMEOUT_S = 30 # exit (let systemd restart) if the broker never connects config_path = sys.argv[1] if len(sys.argv) > 1 else "/etc/van-li3/config.json" with open(config_path) as f: cfg = json.load(f) ADDR = cfg["ble_address"] FFE1 = "0000ffe1-0000-1000-8000-00805f9b34fb" broker = cfg["broker"] mqtt_port = cfg.get("port", 1883) mq_user = cfg["username"] mq_pw = cfg["password"] client_id = cfg.get("client_id", "van-li3-battery") device_id = cfg.get("device_id", "li3_battery") device_name = cfg.get("device_name", "Li3 Battery") discovery_prefix = cfg.get("discovery_prefix", "homeassistant") state_topic = cfg.get("state_topic", f"van/{device_id}/state") publish_interval_s = cfg.get("publish_interval_s", 15) SENSORS = [ # (key, name, unit, device_class, display_precision) -- precision None means # a non-numeric value (hex code, version string, serial number): published # as-is with no unit, no rounding, no state_class. ("voltage", "Pack Voltage", "V", "voltage", 2), ("cell1_voltage", "Cell 1 Voltage", "V", "voltage", 2), ("cell2_voltage", "Cell 2 Voltage", "V", "voltage", 2), ("cell3_voltage", "Cell 3 Voltage", "V", "voltage", 2), ("cell4_voltage", "Cell 4 Voltage", "V", "voltage", 2), ("current", "Current", "A", "current", 2), ("soc", "State of Charge", "%", "battery", 0), ("bms_temperature", "BMS Temperature", "°F", "temperature", 1), ("battery_temperature", "Battery Temperature", "°F", "temperature", 1), # From "&" trace lines (streamed continuously once $traceon is sent) -- # CAN-charger bus fields. This battery has no CAN charger wired up, so # can_charger_voltage/current are observed as a fixed sentinel and # can_charger_status/can_status never change; published anyway since # they're genuine decoded fields. ("remaining_capacity", "Remaining Capacity", "Ah", None, 0), ("remaining_time", "Remaining Time", "min", "duration", 0), ("can_charger_voltage", "CAN Charger Voltage", "V", "voltage", 1), ("can_charger_current", "CAN Charger Current", "A", "current", 1), ("can_charger_status", "CAN Charger Status", None, None, None), ("can_status", "CAN Status", None, None, None), ] # From the "$" info line, sent once by the device right after $info and never # repeated -- published separately (own retained state topic) rather than # merged into the periodic telemetry above. INFO_SENSORS = [ ("total_consumed", "Lifetime Consumed", "Ah", None, 0), ("last_fault_code", "Last Fault Code", None, None, None), ("highest_recorded_temp", "Highest Recorded Temperature", "°F", "temperature", 0), ("lowest_recorded_temp", "Lowest Recorded Temperature", "°F", "temperature", 0), ("firmware_version", "Firmware Version", None, None, None), ("aging_factor_temp", "Aging Factor (Temp)", None, None, 0), ("aging_factor_soc", "Aging Factor (SOC)", None, None, 0), ("serial_number", "Serial Number", None, None, None), ] info_topic = cfg.get("info_topic", f"van/{device_id}/info") # The "status" field (main Cs telemetry line, and identically-coded but # cumulative/latched "last_fault_code" from $info) is a 24-bit flag mask. # Bit meanings pulled from the app's own "advanced" string-array resource # (com.lithionics.bms base.apk, array/advanced -- dumped with aapt since the # app's Kotlin StatusCodeTable class references stale/wrong resource IDs and # can't be trusted). Index 0 = bit 23 (MSB) down to index 23 = bit 0 (LSB); # blank entries are unused bits. Applied to "status" only -- last_fault_code # is a lifetime latch (many bits accumulate over time) and isn't meaningfully # summarized the same way. STATUS_FLAGS = [ "", "BMS Temp High", "Overcurrent State", "Charge OFF", "Aux Input State", "Cell Temp Low", "Cell Temp High", "AGSR State", "Temp Sensor Fault", "CAN Charger Fault", "CAN Charger Present", "AC Power Present", "Contactor Flutter", "Pre-Charge Fault", "Contactor Fault", "Contactor State", "Power Off State", "Battery Protection", "Low Voltage", "Reserve Range", "OptoLoop Open", "NeverDie Reserve", "Charge Detected", "High Voltage", ] def status_value(status): """Parse the raw hex status field to an int; None if unparseable. Single source of truth for both the "is this OK" gate and the flag decode, so the two can't disagree on e.g. "0" vs "000000".""" try: return int(status, 16) except (ValueError, TypeError): return None def decode_status(hex_code): value = status_value(hex_code) if value is None: return hex_code active = [ label for i, label in enumerate(STATUS_FLAGS) if label and (value >> (23 - i)) & 1 ] return ", ".join(active) if active else "OK" def parse_line(line): """Cs-series main telemetry line (no prefix character).""" parts = line.split(",") try: f0 = int(parts[0]) except ValueError: return None if not (101 <= f0 <= 9999): return None # not a Cs-series main telemetry line try: return { "voltage": round(f0 * 0.01, 2), "cell1": round(int(parts[1]) * 0.01, 2), "cell2": round(int(parts[2]) * 0.01, 2), "cell3": round(int(parts[3]) * 0.01, 2), "cell4": round(int(parts[4]) * 0.01, 2), "bms_temp_f": int(parts[5]), "batt_temp_f": int(parts[6]), "current_a": round(int(parts[7]) * 0.1, 1), # reported in 0.1A units "soc_pct": int(parts[8]), "status": parts[9] if len(parts) > 9 else "?", } except (ValueError, IndexError): return None def parse_trace_line(line): """'&' trace line: &,batteryId,remaining,remainingTime,canChargerVoltage, canChargerCurrent,canChargerStatus,canStatus""" parts = line.split(",") try: return { "remaining_capacity": int(parts[2]), "remaining_time": int(parts[3]), "can_charger_voltage": round(int(parts[4]) * 0.1, 1), "can_charger_current": round(int(parts[5]) * 0.1, 1), "can_charger_status": parts[6], "can_status": parts[7], } except (ValueError, IndexError): return None def parse_info_line(line): """'$' info line (response to $info): $,totalConsumed,lastFaultCode, highestRecordedTemp,lowestRecordedTemp,firmwareVersion,agingFactorTemp, agingFactorSoc,serialNumber""" parts = line.split(",") try: return { "total_consumed": int(parts[1]), "last_fault_code": parts[2], "highest_recorded_temp": int(parts[3]), "lowest_recorded_temp": int(parts[4]), "firmware_version": parts[5], "aging_factor_temp": int(parts[6]), "aging_factor_soc": int(parts[7]), "serial_number": parts[8], } except (ValueError, IndexError): return None def _sensor_config(key, name, unit, device_class, precision, topic_state, device_info, expire_after=None): numeric = precision is not None payload = { "name": name, "unique_id": f"{device_id}_{key}", "state_topic": topic_state, "unit_of_measurement": unit, "device_class": device_class, "value_template": ( f"{{{{ value_json.{key} | round({precision}) }}}}" if numeric else f"{{{{ value_json.{key} }}}}" ), "device": device_info, } if expire_after is not None: payload["expire_after"] = expire_after if numeric: payload["suggested_display_precision"] = precision payload["state_class"] = "measurement" return payload def publish_discovery(mqc): device_info = { "identifiers": [device_id], "name": device_name, "manufacturer": "Lithionics", "model": "Li3 BMS", } for key, name, unit, device_class, precision in SENSORS: topic = f"{discovery_prefix}/sensor/{device_id}/{key}/config" payload = _sensor_config( key, name, unit, device_class, precision, state_topic, device_info, expire_after=publish_interval_s * 8, ) mqc.publish(topic, json.dumps(payload), retain=True) status_topic = f"{discovery_prefix}/sensor/{device_id}/status/config" status_payload = { "name": "Status", "unique_id": f"{device_id}_status", "state_topic": state_topic, "value_template": "{{ value_json.status_text }}", "device": device_info, "expire_after": publish_interval_s * 8, } mqc.publish(status_topic, json.dumps(status_payload), retain=True) for key, name, unit, device_class, precision in INFO_SENSORS: topic = f"{discovery_prefix}/sensor/{device_id}/{key}/config" payload = _sensor_config(key, name, unit, device_class, precision, info_topic, device_info) payload["entity_category"] = "diagnostic" mqc.publish(topic, json.dumps(payload), retain=True) def publish_info(mqc, info): """Publish the once-per-connection '$info' fields as their own retained message, separate from the periodic telemetry state.""" mqc.publish(info_topic, json.dumps(info), retain=True) print("published info:", info) def build_state_payload(cs_reading, trace_fields): # A non-zero status means the rest of the Cs line's fields are unreliable # (observed 2026-08-18: status '69' alongside e.g. current=341, soc=340, # cell4_voltage=34013.63) — publish only the status in that case. Trace # ("&" line) fields come from a separate message and are published # regardless. status = cs_reading["status"] # "OK" == integer zero in ANY representation. Gate on the parsed int (not # the raw string) so "0"/"0000"/"000000" all agree with decode_status, and # an unparseable status (e.g. "?") is treated as unreliable and suppressed. value = status_value(status) if value is None or value != 0: payload = {"status": status} else: payload = { "voltage": cs_reading["voltage"], "cell1_voltage": cs_reading["cell1"], "cell2_voltage": cs_reading["cell2"], "cell3_voltage": cs_reading["cell3"], "cell4_voltage": cs_reading["cell4"], "current": cs_reading["current_a"], "soc": cs_reading["soc_pct"], "bms_temperature": cs_reading["bms_temp_f"], "battery_temperature": cs_reading["batt_temp_f"], "status": status, } payload["status_text"] = decode_status(status) payload.update(trace_fields) return payload async def restart_bluetooth_service(): print("scan failed; restarting bluetooth.service") # Run systemctl + the settle delay off the event loop (worker thread) so # they don't block the loop; paho drives its own socket thread regardless. try: await asyncio.to_thread( subprocess.run, ["systemctl", "restart", "bluetooth.service"], check=False, timeout=30, ) except (subprocess.SubprocessError, OSError) as e: print("bluetooth restart failed:", e) await asyncio.sleep(3) async def find_device(): """Scan/connect retry loop. Never gives up; self-heals a wedged bluetoothd discovery state along the way. Returns a found device.""" consecutive_failures = 0 last_bluetooth_restart = 0.0 while True: print("scanning for device...") dev = await BleakScanner.find_device_by_address(ADDR, timeout=20) if dev: return dev consecutive_failures += 1 print(f"device not found in scan (attempt {consecutive_failures})") if consecutive_failures >= STUCK_DISCOVERY_THRESHOLD: now = time.time() if (now - last_bluetooth_restart) >= BLUETOOTH_RESTART_COOLDOWN_S: await restart_bluetooth_service() last_bluetooth_restart = now consecutive_failures = 0 await asyncio.sleep(BLE_RETRY_DELAY_S) async def stream_from_device(dev, mqc): """Connect to `dev` and publish readings until it disconnects.""" buf = "" last_publish = 0.0 latest_cs = {} latest_trace = {} info_published = False def notify_handler(_sender, data): nonlocal buf, info_published buf += data.decode("utf-8", errors="replace") while "\r\n" in buf: line, buf = buf.split("\r\n", 1) line = line.strip() if not line: continue if line.startswith("&"): trace = parse_trace_line(line) if trace: latest_trace.update(trace) elif line.startswith("$"): if not info_published: info = parse_info_line(line) if info: publish_info(mqc, info) info_published = True else: reading = parse_line(line) if reading: latest_cs.update(reading) async with BleakClient(dev, timeout=30) as client: print("connected") await client.start_notify(FFE1, notify_handler) await client.write_gatt_char(FFE1, b"$traceon\r\n", response=False) await asyncio.sleep(2) await client.write_gatt_char(FFE1, b"$info\r\n", response=False) while client.is_connected: await asyncio.sleep(1) now = time.time() if latest_cs and (now - last_publish) >= publish_interval_s: payload = build_state_payload(latest_cs, latest_trace) mqc.publish(state_topic, json.dumps(payload)) # print("published:", payload) last_publish = now print("disconnected") async def main(): mqc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id) mqc.username_pw_set(mq_user, mq_pw) # Reassert HA discovery on every (re)connect. If mosquitto restarts and # loses its retained messages, the whole device disappears from HA until # discovery is re-sent; doing it from here (paho thread) keeps the entities # self-restoring. Idempotent on the very first connect. def _reassert_discovery(client, *args): publish_discovery(client) mqc.on_connect_callback_add(_reassert_discovery) mqc.connect(broker, mqtt_port, keepalive=60) mqc.loop_start() # The one failure that makes this service pointless is "no broker at all": # streaming BLE to a dead broker just wastes power and hides the fault. So # give the connection a bounded startup window and exit (letting systemd # restart us) if it never establishes. A good start's transient drops are # left to paho's built-in auto-reconnect above. deadline = time.monotonic() + MQTT_STARTUP_TIMEOUT_S while not mqc.is_connected() and time.monotonic() < deadline: await asyncio.sleep(1) if not mqc.is_connected(): print("ERROR: MQTT broker %s:%s not reachable; exiting" % (broker, mqtt_port)) mqc.loop_stop() mqc.disconnect() sys.exit(1) publish_discovery(mqc) while True: dev = await find_device() try: await stream_from_device(dev, mqc) except Exception as e: print("connection error:", e) await asyncio.sleep(BLE_RETRY_DELAY_S) if __name__ == "__main__": asyncio.run(main())