Python之python-can

见贤思齐

Python之python-can文章目录Python之python-can1、python-can基本2、python-can安装3、配置3.1代码中直接配置3.2使用配置文件配置3.3使用环境变量配置4、基本使用4.1发送单帧报文到总线4.2发送周期报文到总线4.3接收总线的can报文5、使用Listener和NotifIer6、bus总线设置过滤报文7、提供的脚本7.1can.logger7.2can.player7.3can.viewer7.4can.logconvert1、python-can基本python-can库为Python提供控制器局域网支持，为不同的硬件设备提供通用抽象，以及一套用于在CAN总线上发送和接收消息的实用程序。python-can可以在任何Python运行的地方运行；从CAN的高功率计算机到USB设备，再到运行linux的低功率设备，例如BeagleBone或RaspberryPi。更具体地说，该库的一些示例用途：被动记录CAN总线上发生的情况。例如，使用OBD-II端口监控商用车辆。测试通过CAN交互的硬件。在现代汽车、摩托车、船只甚至轮椅中发现的模块已经使用这个库从Python中测试了组件。在回路中对新的硬件模块或软件算法进行原型设计。轻松与现有总线交互。创建虚拟模块以原型CAN总线通信。2、python-can安装使用pip安装pipinstallpython-can1同时根据使用的硬件设备需要安装对应的驱动等，参照Installation—python-can4.0.0documentation3、配置3.1代码中直接配置can对象公开了一个rc字典，可用于设置interface和channelimportcancan.rc['interface']='neovi'#配置硬件类型can.rc['channel']=2#配置通道，根据具体的硬件，int或者strcan.rc['bitrate']=500000#波特率fromcan.interfaceimportBusbus=Bus()#使用rc字典中的配置实例化can123456789也可以在代码中直接指定接口和通道实例化canimportcanbus=can.interface.Bus(bustype='socketcan',channel='vcan0',bitrate=500000)bus2=bus=can.interface.Bus(bustype='pcan',channel='PCAN_USBBUS1',bitrate=250000)123453.2使用配置文件配置在Linux系统上，配置文件在以下路径中搜索：~/can.conf/etc/can.conf$HOME/.can$HOME/.canrc1234567在Windows系统上，配置文件在以下路径中搜索：%USERPROFILE%/can.confcan.ini（当前工作目录）%APPDATA%/can.inips：%USERPROFILE%目录为C:\Users\用户名%APPDATA%目录为C:\Users\用户名\AppData\Roaming123456789配置文件可设置默认接口和通道如下：[default]interface=channel=bitrate=1234同时也可以添加其他节点如下：[HS]#Allthevaluesfromthe'default'sectionareinheritedchannel=bitrate=[MS]#Allthevaluesfromthe'default'sectionareinheritedchannel=bitrate=12345678910111213使用配置文件的配置实例化canfromcan.interfaceimportBusbus=Bus()#使用默认（default）配置hs_bus=Bus(context='HS')#使用HS节点配置ms_bus=Bus(context='MS')#使用MS节点配置123453.3使用环境变量配置可以设置以下环境变量CAN_INTERFACE设置接口类型CAN_CHANNEL设置通道CAN_BITRATE设置波特率CAN_CONFIGCAN_CONFIG允许使用JSON设置任何总线配置，例如：CAN_CONFIG={"receive_own_messages":true,"fd":true}12345678910114、基本使用4.1发送单帧报文到总线实例化一条bus总线，使用can.Message()类创建一条message，bus调用send函数将创建的message实例发送到总线。importcandefsend_one():#thisusesthedefaultconfiguration(forexamplefromtheconfigfile)#seehttps://python-can.readthedocs.io/en/stable/configuration.html#bus=can.interface.Bus()#Usingspecificbusesworkssimilar:bus=can.interface.Bus(bustype='bmcan',channel=0,bitrate=500000,data_bitrate=2000000,tres=True)#bus=can.interface.Bus(bustype='socketcan',channel='vcan0',bitrate=250000)#bus=can.interface.Bus(bustype='pcan',channel='PCAN_USBBUS1',bitrate=250000)#bus=can.interface.Bus(bustype='ixxat',channel=0,bitrate=250000)#bus=can.interface.Bus(bustype='vector',app_name='CANalyzer',channel=0,bitrate=250000)#...msg=can.Message(arbitration_id=0xc0ffee,data=[0,25,0,1,3,1,4,1],is_extended_id=True,)try:bus.send(msg,timeout=None)print("Messagesenton{}".format(bus.channel_info))exceptcan.CanError:print("MessageNOTsent")bus.shutdown()if__name__=='__main__':send_one()1234567891011121314151617181920212223242526272829304.2发送周期报文到总线以下实例分别演示简单发送周期报文，发送定时的周期报文，和发送周期报文过程中修改发送的内容。注意：在发送定时的周期报文时，达到超时时间任务结束后，task将仍旧由总线跟踪。除非在创建任务时设置参数store_task=False，或者调用stop()函数结束任务task.stop(）。importtimeimportcandefsimple_periodic_send(bus):"""发送一条周期为200ms的周期报文到总线2s后停止发送报文"""print("Startingtosendamessageevery200msfor2s")msg=can.Message(arbitration_id=0x123,data=[1,2,3,4,5,6],is_extended_id=False)task=bus.send_periodic(msg,0.20)assertisinstance(task,can.CyclicSendTaskABC)time.sleep(2)task.stop()print("stoppedcyclicsend")deflimited_periodic_send(bus):"""发送一条有时间限制的周期报文到总线，达到超时时间后，停止任务"""print("Startingtosendamessageevery200msfor1s")msg=can.Message(arbitration_id=0x12345678,data=[0,0,0,0,0,0],is_extended_id=True)task=bus.send_periodic(msg,0.20,1,store_task=False)ifnotisinstance(task,can.LimitedDurationCyclicSendTaskABC):print("Thisinterfacedoesn'tseemtosupporta")task.stop()returntime.sleep(2)deftest_periodic_send_with_modifying_data(bus):"""在发送周期报文过程中修改发送的数据"""print("Startingtosendamessageevery200ms.Initialdataisones")msg=can.Message(arbitration_id=0x0cf02200,data=[1,1,1,1])task=bus.send_periodic(msg,0.20)ifnotisinstance(task,can.ModifiableCyclicTaskABC):print("Thisinterfacedoesn'tseemtosupportmodification")task.stop()returntime.sleep(2)print("Changingdataofrunningtasktobeginwith99")msg.data[0]=0x99task.modify_data(msg)time.sleep(2)task.stop()print("stoppedcyclicsend")print("Changingdataofstoppedtasktosingleffbyte")msg.data=bytearray([0xff])msg.dlc=1task.modify_data(msg)time.sleep(1)print("startingagain")task.start()time.sleep(1)task.stop()print("done")12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758594.3接收总线的can报文以下示例演示使用recv()函数接收来自总线的can报文，然后打印报文。importcandefreceive_all():bus=can.interface.Bus(bustype='ixxat',channel=0,bitrate=500000)print('WaitingforRXCANmessages...')try:whileTrue:msg=bus.recv(1)ifmsgisnotNone:print(msg)exceptKeyboardInterrupt:passif__name__=="__main__":receive_all()1234567891011121314151617185、使用Listener和NotifIerNotifier对象用作总线的消息分发器。Notifier创建一个线程来从总线读取消息并将它们分发给listeners。Listener类是任何希望注册以接收总线上新消息通知的对象的“抽象”基类。通常使用是需继承can.Listener。重写on_message_received()方法，以实现收到特定的消息的处理逻辑。以下示例，实现MyListener处理收到message0x123时，发送message0x456到总线，打印message的回调函数，创建一个日志记录器，添加到listeners列表。实例化一个Notifier对象，添加监听的bus总线，绑定listeners列表。当Notifier对象收到bus总线的任意消息，则会将该消息分发listeners列表中的各个监听器。Notifier对象使用stop()函数以结束监听bus。有一些侦听器python-can已经提供，其中一些允许将消息写入文件，例如can.Logger。importcanfromcan.messageimportMessageclassMyListener(can.Listener):def__init__(self):super(MyListener,self).__init__()defon_message_received(self,msg:Message)->None:"""examplewhenreceivemessage0x123，transmitmessage0x456"""ifmsg.arbitration_id==0x123:transmit_msg=can.Message(arbitration_id=0x456,dlc=8,data=[0for_inrange(8)],is_extended_id=False)bus.send(transmit_msg)defprint_msg(msg):print(msg)if__name__=="__main__":bus=can.interface.Bus('virtual_ch',bustype='virtual')logger=can.Logger("logfile.asc")#savelogtoascfilelisteners=[print_msg,#Callbackfunction,printthereceivedmessageslogger,#savereceivedmessagestoascfileMyListener#mylistener]notifier=can.Notifier(bus,listeners)running=Truewhilerunning:input()running=False#It'simportanttostopthenotifierinordertofinishthewrittingofascfilenotifier.stop()#stopsthebusbus.shutdown()123456789101112131415161718192021222324252627282930313233343536373839404142434445466、bus总线设置过滤报文可以使用set_filters()设置过滤报文，对bus接收到的所有报文进行过滤，返回过滤器匹配的消息。set_filters()参数为一个可迭代的字典，每个字典都包含一个“can_id”、一个“can_mask”和一个可选的“extended”键。[{"can_id":0x11,"can_mask":0x21,"extended":False}]1'运行运行过滤器会匹配&can_mask==can_id&can_mask的所有消息。如果extended参数也设置了，则仅会匹配==extended的消息。否则，它仅根据仲裁ID和掩码匹配每条消息。bus=can.interface.Bus('virtual_ch',bustype='virtual')bus.set_filters([{"can_id":0x123,"can_mask":0xFFFF,"extended":False}])127、提供的脚本python-can已经提供一些可直接使用的脚本参照Scripts—python-can4.0.0documentation7.1can.loggercan.logger用于记录CANlog，将消息打印到标准输出或给定文件可以使用python-mcan.logger或者can_logger.py的命令行方式调用7.2can.playercan.player用于回放记录的can报文日志文件7.3can.viewer一个带有GUI界面的简单的can总线报文消息的查看器7.4can.logconvertcan.logconvert用于日志文件格式的转换，将日志文件从一种格式转换为另一种格式《AUTOSAR谱系分解(ETAS工具链)》之总目录