154 lines
4.8 KiB
Python
154 lines
4.8 KiB
Python
import ctypes
|
|
import os
|
|
import time
|
|
import threading
|
|
|
|
dll_path = os.path.abspath("ControlCANFD.dll")
|
|
canfd = ctypes.cdll.LoadLibrary(dll_path)
|
|
|
|
DEVICE_TYPE = 41
|
|
DEVICE_INDEX = 0
|
|
BAUD_RATE = 500000
|
|
CAN_CHANNEL_0 = 0
|
|
CAN_CHANNEL_1 = 1
|
|
TARGET_ID = 0x11
|
|
|
|
class ZCAN_CHANNEL_INIT_CONFIG(ctypes.Structure):
|
|
_fields_ = [
|
|
("acc_code", ctypes.c_uint),
|
|
("acc_mask", ctypes.c_uint),
|
|
("reserved", ctypes.c_uint),
|
|
("filter", ctypes.c_ubyte),
|
|
("mode", ctypes.c_ubyte),
|
|
("padding1", ctypes.c_ubyte),
|
|
("padding2", ctypes.c_ubyte),
|
|
("baud_rate", ctypes.c_uint),
|
|
]
|
|
|
|
class ZCAN_CAN_FRAME(ctypes.Structure):
|
|
_fields_ = [
|
|
("can_id", ctypes.c_uint),
|
|
("data_len", ctypes.c_ubyte),
|
|
("flags", ctypes.c_ubyte),
|
|
("__res0", ctypes.c_ubyte),
|
|
("__res1", ctypes.c_ubyte),
|
|
("data", ctypes.c_ubyte * 8),
|
|
]
|
|
|
|
class ZCAN_Transmit_Data(ctypes.Structure):
|
|
_fields_ = [
|
|
("frame", ZCAN_CAN_FRAME),
|
|
("transmit_type", ctypes.c_uint),
|
|
]
|
|
|
|
ZCAN_OpenDevice = canfd.ZCAN_OpenDevice
|
|
ZCAN_OpenDevice.argtypes = [ctypes.c_uint, ctypes.c_uint, ctypes.c_uint]
|
|
ZCAN_OpenDevice.restype = ctypes.c_void_p
|
|
|
|
ZCAN_InitCAN = canfd.ZCAN_InitCAN
|
|
ZCAN_InitCAN.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.POINTER(ZCAN_CHANNEL_INIT_CONFIG)]
|
|
ZCAN_InitCAN.restype = ctypes.c_void_p
|
|
|
|
ZCAN_StartCAN = canfd.ZCAN_StartCAN
|
|
ZCAN_StartCAN.argtypes = [ctypes.c_void_p]
|
|
ZCAN_StartCAN.restype = ctypes.c_uint
|
|
|
|
ZCAN_Transmit = canfd.ZCAN_Transmit
|
|
ZCAN_Transmit.argtypes = [ctypes.c_void_p, ctypes.POINTER(ZCAN_Transmit_Data), ctypes.c_uint]
|
|
ZCAN_Transmit.restype = ctypes.c_uint
|
|
|
|
ZCAN_Receive = canfd.ZCAN_Receive
|
|
ZCAN_Receive.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint, ctypes.c_int]
|
|
ZCAN_Receive.restype = ctypes.c_uint
|
|
|
|
def open_can_channel(index):
|
|
dev = ZCAN_OpenDevice(DEVICE_TYPE, DEVICE_INDEX, 0)
|
|
if not dev:
|
|
print(f"[CAN{index}] ❌ Falha ao abrir dispositivo")
|
|
return None, None
|
|
|
|
config = ZCAN_CHANNEL_INIT_CONFIG(
|
|
acc_code=0,
|
|
acc_mask=0xFFFFFFFF,
|
|
reserved=0,
|
|
filter=0,
|
|
mode=0,
|
|
padding1=0,
|
|
padding2=0,
|
|
baud_rate=BAUD_RATE
|
|
)
|
|
|
|
chn = ZCAN_InitCAN(dev, index, ctypes.byref(config))
|
|
if not chn:
|
|
print(f"[CAN{index}] ❌ Falha ao iniciar canal")
|
|
return None, None
|
|
|
|
if ZCAN_StartCAN(chn) != 1:
|
|
print(f"[CAN{index}] ❌ Falha ao iniciar CAN")
|
|
return None, None
|
|
|
|
print(f"[CAN{index}] ✅ Canal iniciado com sucesso")
|
|
return dev, chn
|
|
|
|
def send_can_message(chn, target_id, payload):
|
|
frame = ZCAN_CAN_FRAME()
|
|
frame.can_id = target_id
|
|
frame.data_len = len(payload)
|
|
frame.flags = 0
|
|
frame.__res0 = 0
|
|
frame.__res1 = 0
|
|
frame.data = (ctypes.c_ubyte * 8)(*payload, *([0] * (8 - len(payload))))
|
|
tx = ZCAN_Transmit_Data()
|
|
tx.frame = frame
|
|
tx.transmit_type = 0
|
|
result = ZCAN_Transmit(chn, ctypes.byref(tx), 1)
|
|
print(f"[CAN TX] ID: {target_id:03X} Data: {payload} => {'✅' if result > 0 else '❌'}")
|
|
|
|
def responder_ack_thread(chn_recv, chn_send):
|
|
while True:
|
|
buf = ctypes.create_string_buffer(ctypes.sizeof(ZCAN_CAN_FRAME))
|
|
result = ZCAN_Receive(chn_recv, buf, 1, 100)
|
|
if result > 0:
|
|
frame = ctypes.cast(buf, ctypes.POINTER(ZCAN_CAN_FRAME)).contents
|
|
data = list(frame.data[:frame.data_len])
|
|
print(f"[CAN1 RX] ID: {frame.can_id:03X} Len: {frame.data_len} Data: {data}")
|
|
ack = [0xAA] + data
|
|
send_can_message(chn_send, frame.can_id + 1, ack)
|
|
time.sleep(0.05)
|
|
|
|
def escutar_ack(chn_recv, timeout_ms=1000):
|
|
buf = ctypes.create_string_buffer(ctypes.sizeof(ZCAN_CAN_FRAME))
|
|
start_time = time.time()
|
|
while (time.time() - start_time) * 1000 < timeout_ms:
|
|
result = ZCAN_Receive(chn_recv, buf, 1, 50)
|
|
if result > 0:
|
|
frame = ctypes.cast(buf, ctypes.POINTER(ZCAN_CAN_FRAME)).contents
|
|
data = list(frame.data[:frame.data_len])
|
|
print(f"[CAN0 RX] ID: {frame.can_id:03X} Len: {frame.data_len} Data: {data}")
|
|
if frame.can_id == TARGET_ID + 1:
|
|
print(f"[CAN0 RX] ✅ ACK recebido")
|
|
return
|
|
else:
|
|
print(f"[CAN0 RX] ⚠️ Ignorado (ID inesperado)")
|
|
else:
|
|
print("[CAN0 RX] ... esperando")
|
|
time.sleep(0.05)
|
|
print("[CAN0 RX] ❌ Nenhum ACK recebido")
|
|
|
|
|
|
|
|
# --- Execução principal ---
|
|
dev0, chn0 = open_can_channel(CAN_CHANNEL_0)
|
|
dev1, chn1 = open_can_channel(CAN_CHANNEL_1)
|
|
|
|
if chn0 and chn1:
|
|
# Iniciar thread que escuta e responde
|
|
threading.Thread(target=responder_ack_thread, args=(chn1, chn1), daemon=True).start()
|
|
time.sleep(0.5)
|
|
|
|
# Enviar mensagem do canal 0
|
|
send_can_message(chn0, TARGET_ID, [0x00, 0x01])
|
|
|
|
# Esperar resposta no canal 0
|
|
escutar_ack(chn0)
|