硬件:
canable v2.0(slcan固件,支持canfd)
一个j1900小主机
系统:
Ubuntu24.04(在Ubuntu和xubuntu上都可)
python3.12
用到的Python库:
cantools==39.3.0
mqtt==0.0.1
paho-mqtt==1.6.1
pyserial==3.5
python-can==4.2.0
食用方法:
1、挂载内核模块和apt下载包
挂载内核模块:
sudo modprobe can
sudo modprobe can_raw
sudo modprobe slcan
安装can-utils
sudo apt install can-utils
(玄学但有用,不进行以上操作可能会导致没有数据输出)
安装pip和python-venv
sudo apt install python3-pip
sudo apt install python3-venv
2、激活虚拟环境
直接pip install 在python3.12中会报错:
error: externally-managed-environment
× This environment is externally managed
╰─> To install Python packages system-wide, try apt install
python3-xyz, where xyz is the package you are trying to
install.
If you wish to install a non-Debian-packaged Python package,
create a virtual environment using python3 -m venv path/to/venv.
Then use path/to/venv/bin/python and path/to/venv/bin/pip. Make
sure you have python3-full installed.
If you wish to install a non-Debian packaged Python application,
it may be easiest to use pipx install xyz, which will manage a
virtual environment for you. Make sure you have pipx installed.
See /usr/share/doc/python3.11/README.venv for more information.
note: If you believe this is a mistake, please contact your Python installation or OS distribution provider. You can override this, at the risk of breaking your Python installation or OS, by passing --break-system-packages.
hint: See PEP 668 for the detailed specification.
那就整个venv干:
mkdir -p $HOME/.env && python3 -m venv $HOME/.env/default #以default为例
$HOME/.env/default /bin/python -m pip install -r ./requirements.txt
$HOME/.env/default /bin/python ./test.py
这样python脚本就好跑了。
在此之前,记得给你用的canable给了777:
sudo chmod 777 /dev/ttyACM0
附:python代码:
# This is a sample Python script.
import asyncio
import time
from typing import List
import can
from can.notifier import MessageRecipient
import cantools
import serial.tools.list_ports
import paho.mqtt.client as mqtt
import json
client = mqtt.Client()
db = cantools.database.load_file('.\test.dbc')
def on_connect(client, userdata, flags, rc):
print(f"Connected with result code {rc}")
async def main() -> None:
client.on_connect = on_connect
client.username_pw_set(username="publisher", password="123456")
client.connect("127.0.0.1", 1883, 60)
"""The main function that runs in the loop."""
with can.interface.Bus(bustype='slcan', channel='/dev/ttyACM0', bitrate=500000, ) as bus:
reader = can.AsyncBufferedReader()
listeners: List[MessageRecipient] = [
reader,
]
loop = asyncio.get_running_loop()
notifier = can.Notifier(bus, listeners, loop=loop)
while True:
msg = await reader.get_message()
try:
print(msg)
encodeMsg = db.decode_message(msg.arbitration_id, msg.data)
print(encodeMsg)
sendData={'id':msg.arbitration_id,'data':encodeMsg}
client.publish("/payload",json.dump(encodeMsg))
except Exception as ex:
print(ex)
pass
pass
if __name__ == "__main__":
asyncio.run(main())