agrobot_base/Python/dalybms/tradutor.py

#!/usr/bin/env python3
"""
Parser simples para frames CAN do Daly Smart BMS (protocolo V1.0).

Uso:
    python daly_can_parser.py

Depois ir colando linhas assim (exemplos):

    0x18904001 01 94 00 00 75 30 03 B6
    0x18914001 0F A0 01 0F 72 02
    0x18924001 3C 01 32 02
    0x18934001 00 01 02 05 00 01 86 A0
    0x18944001 10 03 01 01 03 00 00 00
    0x18984001 00 00 00 00 00 00 00 00

Enter vazio ou “q” sai.
"""

from typing import List, Tuple

# ------------- helpers básicos ----------------

def parse_hex_token(tok: str) -> int:
    tok = tok.strip()
    if tok.lower().startswith("0x"):
        tok = tok[2:]
    return int(tok, 16)


def split_id_and_data(line: str) -> Tuple[int, List[int]]:
    """
    Aceita formatos como:
        0x18904001 01 94 00 00 75 30 03 B6
        18904001 01 94 00 00 75 30 03 B6
    Ignora lixo no final.
    """
    parts = [p for p in line.replace(",", " ").split() if p]

    if len(parts) < 9:
        raise ValueError("Esperava ID + 8 bytes de dados (mínimo 9 tokens).")

    can_id = parse_hex_token(parts[0])
    data_bytes = [parse_hex_token(tok) & 0xFF for tok in parts[1:9]]

    return can_id, data_bytes


def get_data_id(can_id: int) -> int:
    """
    No protocolo Daly:
        ID = [Priority][DataID][AddrA][AddrB]
    DataID = segundo byte (bits 16..23)
    """
    return (can_id >> 16) & 0xFF


def u16_be(b_hi: int, b_lo: int) -> int:
    return ((b_hi & 0xFF) << 8) | (b_lo & 0xFF)


def bits_set(byte_val: int) -> List[int]:
    """Retorna lista com índices de bits = 1 (0..7)."""
    return [i for i in range(8) if (byte_val >> i) & 1]


# ------------- parsers por DataID ----------------

def parse_0x90(data: List[int]) -> None:
    """SOC / total voltage / current."""
    total_v_raw = u16_be(data[0], data[1])
    meas_v_raw  = u16_be(data[2], data[3])
    current_raw = u16_be(data[4], data[5])
    soc_raw     = u16_be(data[6], data[7])

    total_v = total_v_raw / 10.0
    meas_v  = meas_v_raw / 10.0
    current = (current_raw - 30000) / 10.0
    soc     = soc_raw / 10.0

    print("== 0x90: Tensão / Corrente / SOC ==")
    print(f"  Tensão total (cumulative): {total_v:.1f} V  (raw={total_v_raw})")
    print(f"  Tensão medida  (gather)  : {meas_v:.1f} V  (raw={meas_v_raw})")
    print(f"  Corrente                : {current:.1f} A  (raw={current_raw})")
    print(f"  SOC                     : {soc:.1f} % (raw={soc_raw})")


def parse_0x91(data: List[int]) -> None:
    """Maximum & minimum cell voltage."""
    vmax_raw = u16_be(data[0], data[1])
    vmax_idx = data[2]
    vmin_raw = u16_be(data[3], data[4])
    vmin_idx = data[5]

    print("== 0x91: Tensão máx/mín de célula ==")
    print(f"  Vmax: {vmax_raw/1000.0:.3f} V  (raw={vmax_raw} mV)  célula #{vmax_idx}")
    print(f"  Vmin: {vmin_raw/1000.0:.3f} V  (raw={vmin_raw} mV)  célula #{vmin_idx}")


def parse_0x92(data: List[int]) -> None:
    """Maximum & minimum temperature."""
    tmax   = data[0] - 40
    tmax_i = data[1]
    tmin   = data[2] - 40
    tmin_i = data[3]

    print("== 0x92: Temperatura máx/mín ==")
    print(f"  T_max: {tmax} °C  (sensor #{tmax_i})")
    print(f"  T_min: {tmin} °C  (sensor #{tmin_i})")


def parse_0x93(data: List[int]) -> None:
    """Charge / discharge MOS status + remaining capacity."""
    state = data[0]
    charge_mos = data[1]
    dischg_mos = data[2]
    life_cycles = data[3]
    remain_mAh = (
        (data[4] << 24)
        | (data[5] << 16)
        | (data[6] << 8)
        | data[7]
    )

    state_map = {0: "parado", 1: "carga", 2: "descarga"}

    print("== 0x93: Estado MOS / capacidade restante ==")
    print(f"  Estado geral         : {state} ({state_map.get(state,'desconhecido')})")
    print(f"  Charge MOS state     : {charge_mos}")
    print(f"  Discharge MOS state  : {dischg_mos}")
    print(f"  Vida BMS (ciclos)    : {life_cycles}")
    print(f"  Capacidade restante  : {remain_mAh} mAh ({remain_mAh/1000.0:.3f} Ah)")


def parse_0x94(data: List[int]) -> None:
    """Status info: nº células, temps, charger/load, DI/DO bits."""
    n_series   = data[0]
    n_temps    = data[1]
    charger    = data[2]
    load       = data[3]
    io_byte    = data[4]

    di_bits = bits_set(io_byte & 0x0F)        # bits 0..3
    do_bits = bits_set((io_byte >> 4) & 0x0F) # bits 4..7

    print("== 0x94: Status geral / IO ==")
    print(f"  Nº células em série : {n_series}")
    print(f"  Nº sensores de temp : {n_temps}")
    print(f"  Carregador conectado: {'sim' if charger else 'não'}")
    print(f"  Load conectado      : {'sim' if load else 'não'}")
    print(f"  DI ativos (0-3)     : {['DI'+str(i+1) for i in di_bits] or ['nenhum']}")
    print(f"  DO ativos (0-3)     : {['DO'+str(i+1) for i in do_bits] or ['nenhum']}")


def parse_0x95(data: List[int]) -> None:
    """
    Cell voltage frames (1-48).
    Byte0 = frame number (0..15), Byte1..6 = 3 tensões de célula (mV), Byte7 reservado.
    Cada célula = 2 bytes, 1 mV.
    """
    frame = data[0]
    cells = []
    for i in range(3):  # 3 células por frame
        hi = data[1 + 2*i]
        lo = data[1 + 2*i + 1]
        raw = u16_be(hi, lo)
        if raw == 0xFFFF or raw == 0:
            continue
        cell_index = frame * 3 + i + 1  # 1-based
        cells.append((cell_index, raw))

    print("== 0x95: Tensões individuais de célula (frame múltiplo) ==")
    print(f"  Frame: {frame}")
    if not cells:
        print("  (sem células válidas neste frame)")
    else:
        for idx, raw in cells:
            print(f"  Célula #{idx:2d}: {raw/1000.0:.3f} V  (raw={raw} mV)")


def parse_0x96(data: List[int]) -> None:
    """
    Cell temperatures frames.
    Byte0 = frame number (0..2), Byte1..7 = temps com offset 40.
    """
    frame = data[0]
    temps = []
    for i in range(1, 8):
        raw = data[i]
        if raw == 0xFF:  # vazio
            continue
        temp_c = raw - 40
        sensor_index = frame * 7 + (i - 1) + 1  # 1-based
        temps.append((sensor_index, temp_c))

    print("== 0x96: Temperaturas individuais (frame múltiplo) ==")
    print(f"  Frame: {frame}")
    if not temps:
        print("  (sem temperaturas válidas neste frame)")
    else:
        for idx, t in temps:
            print(f"  Temp sensor #{idx:2d}: {t} °C")


def parse_0x97(data: List[int]) -> None:
    """
    Balance state 1-48.
    0 = off, 1 = on.
    Bits 0..47 mapeiam células 1..48. 48..63 reservados.
    """
    active_cells = []
    for byte_index, b in enumerate(data):
        for bit in range(8):
            global_bit = byte_index * 8 + bit
            if global_bit >= 48:
                continue
            if (b >> bit) & 1:
                cell_idx = global_bit + 1
                active_cells.append(cell_idx)

    print("== 0x97: Estado de balanceamento das células ==")
    if not active_cells:
        print("  Nenhuma célula está em balanceamento (todas 0).")
    else:
        print(f"  Células em balanceamento ({len(active_cells)}): {active_cells}")


def parse_0x98(data: List[int]) -> None:
    """
    Battery failure status: mapa de flags.
    Imprime apenas os alarmes ativos.
    """
    alarms = []

    # Byte0
    b = data[0]
    if b & (1 << 0): alarms.append("Cell volt high level 1")
    if b & (1 << 1): alarms.append("Cell volt high level 2")
    if b & (1 << 2): alarms.append("Cell volt low level 1")
    if b & (1 << 3): alarms.append("Cell volt low level 2")
    if b & (1 << 4): alarms.append("Sum volt high level 1")
    if b & (1 << 5): alarms.append("Sum volt high level 2")
    if b & (1 << 6): alarms.append("Sum volt low level 1")
    if b & (1 << 7): alarms.append("Sum volt low level 2")

    # Byte1
    b = data[1]
    if b & (1 << 0): alarms.append("Chg temp high level 1")
    if b & (1 << 1): alarms.append("Chg temp high level 2")
    if b & (1 << 2): alarms.append("Chg temp low level 1")
    if b & (1 << 3): alarms.append("Chg temp low level 2")
    if b & (1 << 4): alarms.append("Dischg temp high level 1")
    if b & (1 << 5): alarms.append("Dischg temp high level 2")
    if b & (1 << 6): alarms.append("Dischg temp low level 1")
    if b & (1 << 7): alarms.append("Dischg temp low level 2")

    # Byte2
    b = data[2]
    if b & (1 << 0): alarms.append("Chg overcurrent level 1")
    if b & (1 << 1): alarms.append("Chg overcurrent level 2")
    if b & (1 << 2): alarms.append("Dischg overcurrent level 1")
    if b & (1 << 3): alarms.append("Dischg overcurrent level 2")
    if b & (1 << 4): alarms.append("SOC high level 1")
    if b & (1 << 5): alarms.append("SOC high level 2")
    if b & (1 << 6): alarms.append("SOC low level 1")
    if b & (1 << 7): alarms.append("SOC low level 2")

    # Byte3
    b = data[3]
    if b & (1 << 0): alarms.append("Diff volt level 1")
    if b & (1 << 1): alarms.append("Diff volt level 2")
    if b & (1 << 2): alarms.append("Diff temp level 1")
    if b & (1 << 3): alarms.append("Diff temp level 2")

    # Byte4
    b = data[4]
    if b & (1 << 0): alarms.append("Chg MOS temp high alarm")
    if b & (1 << 1): alarms.append("Dischg MOS temp high alarm")
    if b & (1 << 2): alarms.append("Chg MOS temp sensor error")
    if b & (1 << 3): alarms.append("Dischg MOS temp sensor error")
    if b & (1 << 4): alarms.append("Chg MOS adhesion error")
    if b & (1 << 5): alarms.append("Dischg MOS adhesion error")
    if b & (1 << 6): alarms.append("Chg MOS open circuit error")
    if b & (1 << 7): alarms.append("Dischg MOS open circuit error")

    # Byte5
    b = data[5]
    if b & (1 << 0): alarms.append("AFE collect chip error")
    if b & (1 << 1): alarms.append("Voltage collect dropped")
    if b & (1 << 2): alarms.append("Cell temp sensor error")
    if b & (1 << 3): alarms.append("EEPROM error")
    if b & (1 << 4): alarms.append("RTC error")
    if b & (1 << 5): alarms.append("Precharge failure")
    if b & (1 << 6): alarms.append("Communication failure")
    if b & (1 << 7): alarms.append("Internal communication failure")

    # Byte6
    b = data[6]
    if b & (1 << 0): alarms.append("Current module fault")
    if b & (1 << 1): alarms.append("Sum voltage detect fault")
    if b & (1 << 2): alarms.append("Short circuit protect fault")
    if b & (1 << 3): alarms.append("Low volt forbidden chg fault")

    fault_code = data[7]

    print("== 0x98: Status de falhas / alarmes ==")
    if not alarms and fault_code == 0:
        print("  Nenhum alarme ativo.")
    else:
        print("  Alarmes ativos:")
        for a in alarms:
            print("   - " + a)
        print(f"  Fault code bruto: 0x{fault_code:02X} ({fault_code})")


# ------------- roteador principal ----------------

def parse_daly_frame(can_id: int, data: List[int]) -> None:
    data_id = get_data_id(can_id)
    print(f"\n=== Frame recebido ===")
    print(f"  CAN ID : 0x{can_id:08X}")
    print(f"  DataID : 0x{data_id:02X}")
    print(f"  Data   : {' '.join(f'{b:02X}' for b in data)}")

    if data_id == 0x90:
        parse_0x90(data)
    elif data_id == 0x91:
        parse_0x91(data)
    elif data_id == 0x92:
        parse_0x92(data)
    elif data_id == 0x93:
        parse_0x93(data)
    elif data_id == 0x94:
        parse_0x94(data)
    elif data_id == 0x95:
        parse_0x95(data)
    elif data_id == 0x96:
        parse_0x96(data)
    elif data_id == 0x97:
        parse_0x97(data)
    elif data_id == 0x98:
        parse_0x98(data)
    else:
        print("  DataID ainda não implementado neste parser.")


def main():
    print("Daly BMS CAN parser (teste de bancada)")
    print("Cole uma linha com ID + 8 bytes (ex.: 0x18904001 01 94 00 00 75 30 03 B6)")
    print("Enter vazio ou 'q' para sair.\n")

    while True:
        try:
            line = input("> ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nSaindo.")
            break

        if not line or line.lower() in ("q", "quit", "exit"):
            print("Saindo.")
            break

        try:
            can_id, data = split_id_and_data(line)
            parse_daly_frame(can_id, data)
        except Exception as e:
            print(f"Erro ao processar linha: {e}")


if __name__ == "__main__":
    main()