# 安裝通用 CAN 庫(kù)
pip install python-can

# 安裝 canard（UDS 協(xié)議支持）
pip install canard

import can

# 連接 CAN 接口（以 socketcan 為例）
bus = can.interface.Bus(channel='vcan0', bustype='socketcan')

# 發(fā)送單幀數(shù)據(jù)
msg = can.Message(
    arbitration_id=0x123,  # CAN ID
    data=[0x01, 0x02, 0x03, 0x04],
    is_extended_id=False    # 標(biāo)準(zhǔn)幀
)
bus.send(msg)

# 接收數(shù)據(jù)
for msg in bus:
    print(f"Received: ID={hex(msg.arbitration_id)}, Data={msg.data}")

# 關(guān)閉連接
bus.shutdown()

from canard import Canard
from canard.hw import socketcan

# 加載 DBC 文件
dbc = Canard()
dbc.load_dbc('vehicle.dbc')

# 創(chuàng)建虛擬 CAN 接口
dev = socketcan.SocketCanDev('vcan0')
dev.start()

# 解析接收到的報(bào)文
def on_message(msg):
    decoded = dbc.decode_message(msg.arbitration_id, msg.data)
    print(f"信號(hào)值: {decoded}")

dev.add_listener(on_message)

from canard.proto.uds import UdsClient

# 創(chuàng)建 UDS 客戶端
uds = UdsClient(transport=can.Bus(bustype='socketcan', channel='vcan0'))

# 發(fā)送診斷請(qǐng)求：讀取 ECU 序列號(hào)（服務(wù) 0x22，參數(shù) 0xF189）
response = uds.request([0x22, 0xF1, 0x89])
if response:
    print(f"ECU 序列號(hào): {bytes(response.data[2:]).decode()}")
else:
    print("請(qǐng)求超時(shí)")

def test_door_status():
    bus = can.interface.Bus(channel='vcan0', bustype='socketcan')
    
    # 發(fā)送車門(mén)控制命令（ID=0x200，數(shù)據(jù)=[0x01] 表示開(kāi)門(mén)）
    bus.send(can.Message(arbitration_id=0x200, data=[0x01]))
    
    # 等待并驗(yàn)證車門(mén)狀態(tài)反饋（預(yù)期 ID=0x201，數(shù)據(jù)[0]=0x01）
    for msg in bus:
        if msg.arbitration_id == 0x201:
            assert msg.data[0] == 0x01, "車門(mén)未正確打開(kāi)"
            break
    bus.shutdown()

import pytest

???????@pytest.fixture
def can_bus():
    bus = can.interface.Bus(channel='vcan0', bustype='socketcan')
    yield bus
    bus.shutdown()

def test_engine_rpm(can_bus):
    # 發(fā)送轉(zhuǎn)速請(qǐng)求命令
    can_bus.send(can.Message(arbitration_id=0x700, data=[0x03]))
    
    # 驗(yàn)證響應(yīng)是否在合理范圍（如 800-5000 RPM）
    for msg in can_bus:
        if msg.arbitration_id == 0x701:
            rpm = int.from_bytes(msg.data[2:4], byteorder='big')
            assert 800 <= rpm <= 5000, "轉(zhuǎn)速異常"
            break

# 使用 python-can 發(fā)送 CAN FD 報(bào)文
msg = can.Message(
    arbitration_id=0x123,
    data=[i % 256 for i in range(64)],  # 64 字節(jié)數(shù)據(jù)
    is_fd=True,
    bitrate_switch=True  # 啟用比特率切換
)
bus.send(msg)

# 連接 Vector 硬件（需安裝 XL Driver）
bus = can.interface.Bus(
    bustype='vector',
    channel=1,
    app_name='Python_HIL_Test'
)

# 發(fā)送同步周期性報(bào)文
task = bus.send_periodic(msgs=[msg1, msg2], period=0.1)
task.stop()  # 停止發(fā)送

# 使用 pytest-html 生成測(cè)試報(bào)告
pytest.main(["--html=report.html"])

# 使用 canard 處理 ISO-TP 多幀傳輸
from canard.proto.iso_tp import IsoTpTransport

tp = IsoTpTransport(can_bus, tx_id=0x7E0, rx_id=0x7E8)
data = tp.recv()  # 接收長(zhǎng)數(shù)據(jù)（自動(dòng)重組）