vanlink/ha/custom_components/li3_battery/parser.py

"""Lithionics Li3 BMS wire protocol.

Ported verbatim from vanlink's van-li3-battery (li3/van-li3-battery in the
vanlink repo). 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.
"""
from __future__ import annotations

FFE1_CHAR_UUID = "0000ffe1-0000-1000-8000-00805f9b34fb"

# 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 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