466 lines
14 KiB
Python
466 lines
14 KiB
Python
import ctypes
|
|
import time
|
|
import os
|
|
from typing import List
|
|
|
|
# ========= CONFIG BÁSICA =========
|
|
|
|
DLL_NAME = "ControlCANFD.dll"
|
|
DEVICE_TYPE = 41
|
|
DEVICE_INDEX = 0
|
|
CHANNEL_INDEX = 0 # 0 = CAN1, 1 = CAN2
|
|
|
|
BMS_ADDR = 0x01
|
|
PC_ADDR = 0x40
|
|
|
|
# Baud 250 kbps para Waveshare USB-CAN-FD (RefType=0)
|
|
BAUD_250K = 0x1C0008
|
|
|
|
|
|
# ========= ESTRUTURAS DA DLL =========
|
|
|
|
class VCI_INIT_CONFIG(ctypes.Structure):
|
|
_fields_ = [
|
|
("AccCode", ctypes.c_uint),
|
|
("AccMask", ctypes.c_uint),
|
|
("Reserved", ctypes.c_uint),
|
|
("Filter", ctypes.c_ubyte),
|
|
("Timing0", ctypes.c_ubyte),
|
|
("Timing1", ctypes.c_ubyte),
|
|
("Mode", ctypes.c_ubyte),
|
|
]
|
|
|
|
|
|
class VCI_CAN_OBJ(ctypes.Structure):
|
|
_fields_ = [
|
|
("ID", ctypes.c_uint),
|
|
("TimeStamp", ctypes.c_uint),
|
|
("TimeFlag", ctypes.c_ubyte),
|
|
("SendType", ctypes.c_ubyte),
|
|
("RemoteFlag", ctypes.c_ubyte),
|
|
("ExternFlag", ctypes.c_ubyte),
|
|
("DataLen", ctypes.c_ubyte),
|
|
("Data", ctypes.c_ubyte * 8),
|
|
("Reserved", ctypes.c_ubyte * 3),
|
|
]
|
|
|
|
|
|
# ========= HELPERS DE PROTOCOLO DALY =========
|
|
|
|
def u16_be(hi: int, lo: int) -> int:
|
|
return ((hi & 0xFF) << 8) | (lo & 0xFF)
|
|
|
|
|
|
def bits_set(byte_val: int):
|
|
return [i for i in range(8) if (byte_val >> i) & 1]
|
|
|
|
|
|
def get_data_id_from_can_id(can_id: int) -> int:
|
|
"""
|
|
Protocolo Daly:
|
|
ID (29 bits) = [Priority][DataID][AddrA][AddrB]
|
|
DataID está no segundo byte (bits 16..23)
|
|
"""
|
|
return (can_id >> 16) & 0xFF
|
|
|
|
|
|
def make_request_id(data_id: int) -> int:
|
|
"""
|
|
PC -> BMS:
|
|
0x18 [DataID] [BMS_ADDR] [PC_ADDR]
|
|
Ex: DataID=0x90 => 0x18900140
|
|
"""
|
|
return (0x18 << 24) | (data_id << 16) | (BMS_ADDR << 8) | PC_ADDR
|
|
|
|
|
|
def expected_response_id(data_id: int) -> int:
|
|
"""
|
|
BMS -> PC:
|
|
0x18 [DataID] [PC_ADDR] [BMS_ADDR]
|
|
Ex: DataID=0x90 => 0x18904001
|
|
"""
|
|
return (0x18 << 24) | (data_id << 16) | (PC_ADDR << 8) | BMS_ADDR
|
|
|
|
|
|
# ========= PARSERS POR DATAID =========
|
|
|
|
def parse_0x90(data: List[int]) -> None:
|
|
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:
|
|
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:
|
|
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:
|
|
state = data[0]
|
|
charge_mos = data[1]
|
|
dischg_mos = data[2]
|
|
life_raw = 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" BMS life (raw, 'ciclos'): {life_raw}")
|
|
print(f" Capacidade restante : {remain_mAh} mAh ({remain_mAh/1000.0:.3f} Ah)")
|
|
|
|
|
|
def parse_0x94(data: List[int]) -> None:
|
|
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)
|
|
do_bits = bits_set((io_byte >> 4) & 0x0F)
|
|
|
|
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:
|
|
frame = data[0]
|
|
cells = []
|
|
for i in range(3):
|
|
hi = data[1 + 2*i]
|
|
lo = data[1 + 2*i + 1]
|
|
raw = u16_be(hi, lo)
|
|
if raw == 0 or raw == 0xFFFF:
|
|
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:
|
|
frame = data[0]
|
|
temps = []
|
|
for i in range(1, 8):
|
|
raw = data[i]
|
|
if raw == 0xFF:
|
|
continue
|
|
temp_c = raw - 40
|
|
sensor_index = frame * 7 + (i - 1) + 1
|
|
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:
|
|
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:
|
|
alarms = []
|
|
|
|
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")
|
|
|
|
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")
|
|
|
|
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")
|
|
|
|
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")
|
|
|
|
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")
|
|
|
|
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")
|
|
|
|
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})")
|
|
|
|
|
|
def parse_daly_frame(can_id: int, data: List[int]) -> None:
|
|
data_id = get_data_id_from_can_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.")
|
|
|
|
|
|
# ========= INICIALIZAÇÃO DO CAN =========
|
|
|
|
def init_can():
|
|
dll = ctypes.cdll.LoadLibrary(os.path.abspath(DLL_NAME))
|
|
|
|
ret = dll.VCI_OpenDevice(DEVICE_TYPE, DEVICE_INDEX, 0)
|
|
if ret != 1:
|
|
raise RuntimeError("Falha ao abrir dispositivo CAN")
|
|
|
|
baud = ctypes.c_uint(BAUD_250K)
|
|
ret = dll.VCI_SetReference(DEVICE_TYPE, DEVICE_INDEX, CHANNEL_INDEX, 0, ctypes.byref(baud))
|
|
if ret != 1:
|
|
raise RuntimeError("Falha ao configurar baudrate 250 kbps")
|
|
|
|
cfg = VCI_INIT_CONFIG()
|
|
cfg.AccCode = 0
|
|
cfg.AccMask = 0xFFFFFFFF
|
|
cfg.Reserved = 0
|
|
cfg.Filter = 0
|
|
cfg.Timing0 = 0
|
|
cfg.Timing1 = 0
|
|
cfg.Mode = 0
|
|
|
|
ret = dll.VCI_InitCAN(DEVICE_TYPE, DEVICE_INDEX, CHANNEL_INDEX, ctypes.byref(cfg))
|
|
if ret != 1:
|
|
raise RuntimeError("Falha ao inicializar CAN")
|
|
|
|
ret = dll.VCI_StartCAN(DEVICE_TYPE, DEVICE_INDEX, CHANNEL_INDEX)
|
|
if ret != 1:
|
|
raise RuntimeError("Falha ao iniciar CAN")
|
|
|
|
dll.VCI_ClearBuffer(DEVICE_TYPE, DEVICE_INDEX, CHANNEL_INDEX)
|
|
|
|
return dll
|
|
|
|
|
|
# ========= ENVIO + RECEPÇÃO PARA UM DATAID =========
|
|
|
|
def send_and_receive(dll, data_id: int, timeout_s: float = 0.5):
|
|
req_id = make_request_id(data_id)
|
|
resp_id_expect = expected_response_id(data_id)
|
|
|
|
send_obj = VCI_CAN_OBJ()
|
|
send_obj.ID = req_id
|
|
send_obj.SendType = 0
|
|
send_obj.RemoteFlag = 0
|
|
send_obj.ExternFlag = 1
|
|
send_obj.TimeFlag = 0
|
|
send_obj.DataLen = 8
|
|
for i in range(8):
|
|
send_obj.Data[i] = 0x00
|
|
|
|
ret = dll.VCI_Transmit(
|
|
DEVICE_TYPE,
|
|
DEVICE_INDEX,
|
|
CHANNEL_INDEX,
|
|
ctypes.byref(send_obj),
|
|
1
|
|
)
|
|
print(
|
|
f"\n📤 Enviando DataID=0x{data_id:02X} "
|
|
f"ID=0x{req_id:08X} DLC=8 Data=00 00 00 00 00 00 00 00 "
|
|
f"(ret={ret})"
|
|
)
|
|
|
|
recv_buffer = (VCI_CAN_OBJ * 50)()
|
|
start = time.time()
|
|
while time.time() - start < timeout_s:
|
|
count = dll.VCI_Receive(
|
|
DEVICE_TYPE,
|
|
DEVICE_INDEX,
|
|
CHANNEL_INDEX,
|
|
ctypes.byref(recv_buffer),
|
|
50,
|
|
50
|
|
)
|
|
if count > 0:
|
|
for i in range(count):
|
|
obj = recv_buffer[i]
|
|
if obj.DataLen == 0:
|
|
continue
|
|
if obj.ID != resp_id_expect:
|
|
# outros frames no barramento, se existirem
|
|
continue
|
|
data = [obj.Data[j] for j in range(obj.DataLen)]
|
|
parse_daly_frame(obj.ID, data)
|
|
return
|
|
time.sleep(0.05)
|
|
|
|
print(f"⚠️ Sem resposta do BMS para DataID 0x{data_id:02X} em {timeout_s:.1f}s.")
|
|
|
|
|
|
# ========= MAIN INTERATIVO =========
|
|
|
|
def main():
|
|
dll = init_can()
|
|
print("✅ Dispositivo aberto, CAN iniciado em 250 kbps.\n")
|
|
print("Daly BMS console:")
|
|
print(" Digite o DataID (em hex) que deseja ler (90..98),")
|
|
print(" ou 'q' para sair.")
|
|
print("Exemplos: 90, 91, 92, 93, 94, 95, 96, 97, 98\n")
|
|
|
|
try:
|
|
while True:
|
|
s = input("DataID (hex) > ").strip()
|
|
if not s:
|
|
continue
|
|
if s.lower() in ("q", "quit", "exit"):
|
|
break
|
|
|
|
try:
|
|
if s.lower().startswith("0x"):
|
|
s = s[2:]
|
|
data_id = int(s, 16)
|
|
except ValueError:
|
|
print("⚠️ Valor inválido. Digite algo como 90, 91, 98, ou 'q' pra sair.")
|
|
continue
|
|
|
|
if not (0x90 <= data_id <= 0x98):
|
|
print("⚠️ Este console está preparado para 0x90 a 0x98.")
|
|
continue
|
|
|
|
send_and_receive(dll, data_id, timeout_s=0.8)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n⏹ Interrompido pelo usuário.")
|
|
finally:
|
|
# se quiser, poderia chamar VCI_CloseDevice aqui
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|