473 lines
17 KiB
Python
473 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Python DeviceNet从站脚本 - 基于Linux SocketCAN实现
|
||
|
||
此脚本实现了一个简单的DeviceNet从站设备，使用Python标准库通过SocketCAN接口与DeviceNet网络通信。
|
||
|
||
功能特点:
|
||
1. 纯Python实现，无需安装外部库，仅使用Python标准库
|
||
2. 支持SocketCAN接口通信
|
||
3. 实现基本的DeviceNet协议功能:
|
||
- 身份标识对象(Identity Object)属性获取
|
||
- 显式消息请求处理
|
||
- 基本I/O消息支持
|
||
4. 支持通过命令行配置MAC ID和CAN接口名称
|
||
5. 优雅的信号处理和资源释放
|
||
|
||
使用方法:
|
||
# 使用默认参数运行 (MAC ID: 10, 接口: can0)
|
||
python python_devicenet_slave.py
|
||
|
||
# 指定MAC ID和CAN接口
|
||
python python_devicenet_slave.py --mac-id 5 --interface can1
|
||
|
||
# 查看帮助信息
|
||
python python_devicenet_slave.py --help
|
||
|
||
注意事项:
|
||
1. 此脚本需要在支持SocketCAN的Linux环境下运行
|
||
2. 运行前确保已正确配置CAN接口，例如:
|
||
sudo ip link set can0 type can bitrate 125000
|
||
sudo ip link set up can0
|
||
3. 需要适当的权限访问CAN设备接口
|
||
4. 这是一个基础实现，仅支持部分DeviceNet功能，可根据需要扩展
|
||
|
||
作者: DeviceNet Python Implementation
|
||
版本: 1.0
|
||
"""
|
||
|
||
import socket
|
||
import struct
|
||
import fcntl
|
||
import time
|
||
import signal
|
||
import sys
|
||
|
||
# SocketCAN相关常量
|
||
CAN_RAW = 1
|
||
SOL_CAN_RAW = 1
|
||
CAN_RAW_FILTER = 1
|
||
|
||
# DeviceNet协议常量
|
||
# 报文组定义
|
||
DEVICENET_GROUP_1 = 1 # 组1：I/O报文（主站→从站）
|
||
DEVICENET_GROUP_2 = 2 # 组2：显式报文（请求和响应）
|
||
DEVICENET_GROUP_3 = 3 # 组3：I/O报文（从站→主站）
|
||
DEVICENET_GROUP_4 = 4 # 组4：主站报文
|
||
|
||
# 服务代码定义
|
||
SERVICE_GET_ATTRIBUTE_SINGLE = 0x0E # 获取单个属性
|
||
SERVICE_SET_ATTRIBUTE_SINGLE = 0x10 # 设置单个属性
|
||
SERVICE_GET_ATTRIBUTE_ALL = 0x01 # 获取所有属性
|
||
SERVICE_RESET = 0x40 # 重置
|
||
|
||
# 对象类标识符
|
||
OBJECT_NULL = 0x00 # 空对象
|
||
OBJECT_IDENTITY = 0x01 # 身份标识对象
|
||
OBJECT_MESSAGE_ROUTER = 0x02 # 消息路由器
|
||
OBJECT_ASSEMBLY = 0x04 # 装配对象
|
||
OBJECT_CONNECTION_MANAGER = 0x06 # 连接管理器
|
||
OBJECT_SDM = 0x0F # 序列化数据管理器
|
||
OBJECT_GROUP_2_OBJECT = 0x20 # 组2对象
|
||
|
||
# 响应状态码
|
||
STATUS_SUCCESS = 0x00 # 成功
|
||
STATUS_OBJECT_DOES_NOT_EXIST = 0x01 # 对象不存在
|
||
STATUS_OBJECT_SPECIFIC_ERROR = 0x02 # 对象特定错误
|
||
STATUS_SERVICE_NOT_SUPPORTED = 0x03 # 服务不支持
|
||
STATUS_INVALID_LENGTH = 0x04 # 无效长度
|
||
STATUS_INVALID_DATA = 0x05 # 无效数据
|
||
|
||
# 定义CAN帧结构体格式（与Linux内核定义匹配）
|
||
class CanFrame:
|
||
"""CAN帧数据结构"""
|
||
def __init__(self, can_id=0, dlc=0, data=b''):
|
||
self.can_id = can_id # CAN ID
|
||
self.dlc = dlc # 数据长度码
|
||
self.data = data # 数据内容
|
||
|
||
def __str__(self):
|
||
return f"CAN ID: 0x{self.can_id:X}, DLC: {self.dlc}, Data: {self.data.hex()}"
|
||
|
||
# SocketCAN接口类
|
||
class SocketCANInterface:
|
||
"""SocketCAN接口封装类"""
|
||
def __init__(self, interface='can0'):
|
||
self.interface = interface
|
||
self.socket_fd = None
|
||
self.is_open = False
|
||
|
||
def open(self):
|
||
"""打开并配置SocketCAN接口"""
|
||
try:
|
||
# 创建原始CAN套接字
|
||
self.socket_fd = socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW)
|
||
|
||
# 获取CAN接口索引
|
||
ifreq = struct.pack('16sI', self.interface.encode('utf-8'), 0)
|
||
try:
|
||
ifr = fcntl.ioctl(self.socket_fd, 0x8933, "can0") # SIOCGIFINDEX
|
||
except IOError as e:
|
||
print(f"错误: 无法找到CAN接口 {self.interface}: {e}")
|
||
self.socket_fd.close()
|
||
return False
|
||
|
||
# 提取接口索引
|
||
#ifindex = struct.unpack('16sI', ifr)[1]
|
||
|
||
# 绑定套接字到CAN接口
|
||
addr = ("can0",)
|
||
|
||
try:
|
||
self.socket_fd.bind(addr)
|
||
except OSError as e:
|
||
print(f"错误: 无法绑定到CAN接口 {self.interface}: {e}")
|
||
self.socket_fd.close()
|
||
return False
|
||
|
||
self.is_open = True
|
||
print(f"成功打开CAN接口: {self.interface}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"错误: 初始化SocketCAN时出错: {e}")
|
||
if self.socket_fd:
|
||
self.socket_fd.close()
|
||
return False
|
||
|
||
def close(self):
|
||
"""关闭SocketCAN接口"""
|
||
if self.socket_fd:
|
||
self.socket_fd.close()
|
||
self.socket_fd = None
|
||
self.is_open = False
|
||
print(f"已关闭CAN接口: {self.interface}")
|
||
|
||
def send_frame(self, frame):
|
||
"""发送CAN帧"""
|
||
if not self.is_open:
|
||
print("错误: 套接字未打开")
|
||
return False
|
||
|
||
try:
|
||
# 构建CAN帧格式 (can_id[4字节] + can_dlc[1字节] + data[8字节])
|
||
can_frame = struct.pack("=IB3x8s", frame.can_id, frame.dlc, frame.data.ljust(8, b'\x00'))
|
||
self.socket_fd.send(can_frame)
|
||
return True
|
||
except Exception as e:
|
||
print(f"错误: 发送CAN帧失败: {e}")
|
||
return False
|
||
|
||
def receive_frame(self, timeout=0.1):
|
||
"""接收CAN帧，支持超时设置"""
|
||
if not self.is_open:
|
||
print("错误: 套接字未打开")
|
||
return None
|
||
|
||
try:
|
||
# 设置接收超时
|
||
self.socket_fd.settimeout(timeout)
|
||
|
||
# 接收CAN帧
|
||
data = self.socket_fd.recv(16) # CAN帧最大16字节
|
||
|
||
# 解析CAN帧
|
||
can_id, can_dlc, frame_data = struct.unpack("=IB3x8s", data)
|
||
|
||
# 提取有效数据
|
||
frame_data = frame_data[:can_dlc]
|
||
|
||
return CanFrame(can_id=can_id, dlc=can_dlc, data=frame_data)
|
||
|
||
except socket.timeout:
|
||
return None # 超时
|
||
except Exception as e:
|
||
print(f"错误: 接收CAN帧失败: {e}")
|
||
return None
|
||
|
||
class DeviceNetSlave:
|
||
"""DeviceNet从站设备类"""
|
||
def __init__(self, mac_id, interface='can0'):
|
||
"""初始化DeviceNet从站设备"""
|
||
# 基本配置
|
||
self.mac_id = mac_id
|
||
self.interface = interface
|
||
self.can = SocketCANInterface(interface)
|
||
|
||
# 身份标识对象属性
|
||
self.identity = {
|
||
'vendor_id': 0x00000001, # 供应商ID (示例值)
|
||
'device_type': 0x00000001, # 设备类型 (示例值: 通用设备)
|
||
'product_code': 0x00000001, # 产品代码 (示例值)
|
||
'revision_major': 0x01, # 主版本号
|
||
'revision_minor': 0x00, # 次版本号
|
||
'status': 0x00000001, # 状态字 (正常运行)
|
||
'serial_number': 0x00000001, # 序列号
|
||
'product_name': "Python DeviceNet Slave" # 产品名称
|
||
}
|
||
|
||
# I/O数据缓冲区
|
||
self.input_data = bytearray(8) # 从站→主站的输入数据
|
||
self.output_data = bytearray(8) # 主站→从站的输出数据
|
||
|
||
def start(self):
|
||
"""启动从站设备"""
|
||
if not self.can.open():
|
||
return False
|
||
|
||
print(f"DeviceNet从站已启动，MAC ID = {self.mac_id}")
|
||
return True
|
||
|
||
def stop(self):
|
||
"""停止从站设备"""
|
||
self.can.close()
|
||
|
||
def parse_can_id(self, can_id):
|
||
"""解析CAN ID，提取DeviceNet协议信息"""
|
||
group = (can_id >> 9) & 0x07 # 提取组号 (3位)
|
||
message_id = (can_id >> 6) & 0x07 # 提取报文ID (3位)
|
||
mac_id = can_id & 0x3F # 提取MAC ID (6位)
|
||
|
||
return {
|
||
'group': group,
|
||
'message_id': message_id,
|
||
'mac_id': mac_id
|
||
}
|
||
|
||
def create_response_can_id(self, request_can_id):
|
||
"""根据请求CAN ID创建响应CAN ID"""
|
||
# 对于组2报文，响应的报文ID是请求的报文ID + 1
|
||
parsed = self.parse_can_id(request_can_id)
|
||
response_can_id = (DEVICENET_GROUP_2 << 9) | ((parsed['message_id'] + 1) << 6) | self.mac_id
|
||
return response_can_id
|
||
|
||
def process_explicit_request(self, frame):
|
||
"""处理显式请求报文"""
|
||
# 解析显式请求
|
||
if frame.dlc < 5:
|
||
print(f"错误: 显式请求报文长度不足: {frame.dlc}")
|
||
return None
|
||
|
||
# 显式请求格式: [分段标志, 服务代码, 类ID, 实例ID, 属性ID, ...]
|
||
segment_flag = frame.data[0]
|
||
service_code = frame.data[1]
|
||
class_id = frame.data[2]
|
||
instance_id = frame.data[3]
|
||
attribute_id = frame.data[4]
|
||
|
||
print(f"收到显式请求 - 服务: 0x{service_code:02X}, 类: 0x{class_id:02X}, "
|
||
f"实例: {instance_id}, 属性: {attribute_id}")
|
||
|
||
# 创建响应帧
|
||
response_id = self.create_response_can_id(frame.can_id)
|
||
response_data = bytearray(8)
|
||
response_data[0] = 0x00 # 不分段
|
||
response_data[1] = service_code | 0x80 # 响应服务代码
|
||
response_data[2] = STATUS_SUCCESS # 默认成功状态
|
||
|
||
# 处理获取属性请求
|
||
if service_code == SERVICE_GET_ATTRIBUTE_SINGLE:
|
||
if class_id == OBJECT_IDENTITY and instance_id == 1:
|
||
# 处理身份标识对象属性
|
||
if attribute_id == 1: # Vendor ID
|
||
# 将Vendor ID作为4字节值放入响应
|
||
vendor_id = self.identity['vendor_id']
|
||
response_data[3] = (vendor_id >> 24) & 0xFF
|
||
response_data[4] = (vendor_id >> 16) & 0xFF
|
||
response_data[5] = (vendor_id >> 8) & 0xFF
|
||
response_data[6] = vendor_id & 0xFF
|
||
elif attribute_id == 2: # Device Type
|
||
device_type = self.identity['device_type']
|
||
response_data[3] = (device_type >> 24) & 0xFF
|
||
response_data[4] = (device_type >> 16) & 0xFF
|
||
response_data[5] = (device_type >> 8) & 0xFF
|
||
response_data[6] = device_type & 0xFF
|
||
elif attribute_id == 3: # Product Code
|
||
product_code = self.identity['product_code']
|
||
response_data[3] = (product_code >> 24) & 0xFF
|
||
response_data[4] = (product_code >> 16) & 0xFF
|
||
response_data[5] = (product_code >> 8) & 0xFF
|
||
response_data[6] = product_code & 0xFF
|
||
elif attribute_id == 4: # Revision
|
||
response_data[3] = 0
|
||
response_data[4] = self.identity['revision_major']
|
||
response_data[5] = 0
|
||
response_data[6] = self.identity['revision_minor']
|
||
elif attribute_id == 5: # Status
|
||
status = self.identity['status']
|
||
response_data[3] = (status >> 24) & 0xFF
|
||
response_data[4] = (status >> 16) & 0xFF
|
||
response_data[5] = (status >> 8) & 0xFF
|
||
response_data[6] = status & 0xFF
|
||
elif attribute_id == 6: # Serial Number
|
||
serial = self.identity['serial_number']
|
||
response_data[3] = (serial >> 24) & 0xFF
|
||
response_data[4] = (serial >> 16) & 0xFF
|
||
response_data[5] = (serial >> 8) & 0xFF
|
||
response_data[6] = serial & 0xFF
|
||
elif attribute_id == 7: # Product Name (前5个字符)
|
||
product_name = self.identity['product_name'].encode('ascii', 'replace')
|
||
for i in range(min(5, len(product_name))):
|
||
response_data[3 + i] = product_name[i]
|
||
else:
|
||
response_data[2] = STATUS_OBJECT_SPECIFIC_ERROR
|
||
print(f"不支持的属性ID: {attribute_id}")
|
||
else:
|
||
response_data[2] = STATUS_OBJECT_DOES_NOT_EXIST
|
||
print(f"不支持的对象类或实例: 类=0x{class_id:02X}, 实例={instance_id}")
|
||
else:
|
||
response_data[2] = STATUS_SERVICE_NOT_SUPPORTED
|
||
print(f"不支持的服务代码: 0x{service_code:02X}")
|
||
|
||
# 创建响应帧
|
||
response_frame = CanFrame(
|
||
can_id=response_id,
|
||
dlc=8,
|
||
data=response_data
|
||
)
|
||
|
||
return response_frame
|
||
|
||
def process_io_message(self, frame):
|
||
"""处理I/O消息帧"""
|
||
parsed = self.parse_can_id(frame.can_id)
|
||
|
||
# 处理来自主站的输出数据 (组1) - 简化版
|
||
if parsed['group'] == DEVICENET_GROUP_1:
|
||
# 将收到的帧数据复制到输出缓冲区
|
||
data_len = min(frame.dlc, len(self.output_data))
|
||
for i in range(data_len):
|
||
self.output_data[i] = frame.data[i]
|
||
print(f"收到I/O输出数据: {self.output_data[:data_len].hex()}")
|
||
|
||
# 可选：生成I/O输入响应
|
||
response_id = (DEVICENET_GROUP_1 << 9) | (0x01 << 6) | self.mac_id
|
||
response_frame = CanFrame(
|
||
can_id=response_id,
|
||
dlc=len(self.input_data),
|
||
data=self.input_data
|
||
)
|
||
return response_frame
|
||
|
||
return None
|
||
|
||
def process_can_frame(self, frame):
|
||
"""处理接收到的CAN帧"""
|
||
parsed = self.parse_can_id(frame.can_id)
|
||
print(parsed)
|
||
# 检查是否是发给本设备的帧
|
||
if parsed['mac_id'] != self.mac_id:
|
||
return None
|
||
|
||
# 根据帧组类型进行处理
|
||
if parsed['group'] == DEVICENET_GROUP_2:
|
||
# 组2：显式消息
|
||
return self.process_explicit_request(frame)
|
||
elif parsed['group'] == DEVICENET_GROUP_1:
|
||
# 组1：I/O消息
|
||
return self.process_io_message(frame)
|
||
else:
|
||
print(f"未处理的帧组类型: {parsed['group']}")
|
||
return None
|
||
|
||
def receive_and_process_frame(self):
|
||
"""接收并处理CAN帧"""
|
||
# 接收CAN帧
|
||
frame = self.can.receive_frame()
|
||
if frame is None:
|
||
return False
|
||
|
||
# 处理接收到的帧
|
||
response_frame = self.process_can_frame(frame)
|
||
|
||
# 发送响应帧（如果有）
|
||
if response_frame:
|
||
self.can.send_frame(response_frame)
|
||
print(f"已发送响应帧 - ID: 0x{response_frame.can_id:03X}")
|
||
|
||
return True
|
||
|
||
|
||
# 全局变量，用于控制主循环
|
||
exit_flag = False
|
||
|
||
def signal_handler(sig, frame):
|
||
"""信号处理函数，用于优雅退出程序"""
|
||
global exit_flag
|
||
print("收到退出信号，准备退出...")
|
||
exit_flag = True
|
||
|
||
|
||
def main(mac_id=10, interface='can0'):
|
||
"""主函数"""
|
||
import signal
|
||
|
||
# 设置信号处理
|
||
signal.signal(signal.SIGINT, signal_handler)
|
||
print("DeviceNet从站程序启动。按Ctrl+C退出。")
|
||
|
||
# 创建从站设备
|
||
slave = DeviceNetSlave(mac_id, interface)
|
||
|
||
# 启动从站
|
||
if not slave.start():
|
||
print(f"错误: 无法启动DeviceNet从站，接口可能不存在: {interface}")
|
||
return 1
|
||
|
||
try:
|
||
# 主循环
|
||
print("进入主循环，等待并处理CAN帧...")
|
||
while not exit_flag:
|
||
# 尝试接收并处理CAN帧（非阻塞）
|
||
slave.receive_and_process_frame()
|
||
|
||
# 短暂休眠以减少CPU使用
|
||
import time
|
||
time.sleep(0.001) # 1ms延迟
|
||
|
||
except Exception as e:
|
||
print(f"发生异常: {e}")
|
||
finally:
|
||
# 确保资源被释放
|
||
print("关闭DeviceNet从站...")
|
||
slave.stop()
|
||
|
||
print("DeviceNet从站程序已退出。")
|
||
return 0
|
||
|
||
|
||
# 程序入口点
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
|
||
# 创建参数解析器
|
||
parser = argparse.ArgumentParser(
|
||
description='DeviceNet从站程序 - 基于Python的SocketCAN实现',
|
||
formatter_class=argparse.ArgumentDefaultsHelpFormatter
|
||
)
|
||
|
||
# 添加MAC ID参数
|
||
parser.add_argument(
|
||
'-m', '--mac-id',
|
||
type=int,
|
||
default=10,
|
||
choices=range(0, 64),
|
||
help='DeviceNet从站MAC ID (0-63)'
|
||
)
|
||
|
||
# 添加CAN接口参数
|
||
parser.add_argument(
|
||
'-i', '--interface',
|
||
type=str,
|
||
default='can0',
|
||
help='CAN接口名称 (例如: can0)'
|
||
)
|
||
|
||
# 解析命令行参数
|
||
args = parser.parse_args()
|
||
|
||
# 调用主函数，传入解析得到的参数
|
||
exit_code = main(mac_id=args.mac_id, interface=args.interface)
|
||
import sys
|
||
sys.exit(exit_code) |