python 玩转 modbus
modbus_tk :完整modbus协议栈的实现,支持modbus tcp/rtu{1.提供slave模拟器,即模拟modbus server:502), web-based hmi master支持}
modbus协议举例简解:
例:发送:01 03 00 10 00 04 crc1 crc2
01 地址
03 功能码(读)
00 10 起始地址
00 04 起始地址往后多少位
crc1 crc2 CRC校验码
接受:01 03 08 xx1 xx2 xx3 xx4 xx5 xx6 xx7 xx8 crc1 crc2
01 03:与发送一致
08: 数据的长度
xx1 xx2 xx3 xx4 xx5 xx6 xx7 xx8:读取的长度*2
crc1 crc2:CRC校验码
python 实现modbus RTU slave
import sys
import modbus_tk
import struct
import modbus_tk.defines as cst
from modbus_tk import modbus_rtu
import time
import serial
import threading
PORT = 'com3'
#PORT = '/dev/ptyp5'
def main():
"""main"""
global slave_1
logger = modbus_tk.utils.create_logger(name="console", record_format="%(message)s")
#Create the server
server = modbus_rtu.RtuServer(serial.Serial(PORT))
try:
logger.info("running...")
logger.info("enter 'quit' for closing the server")
server.start()
slave_1 = server.add_slave(1)
slave_1.add_block('0', cst.HOLDING_REGISTERS, 0, 100)
t = threading.Thread(target=updateData)
t.setDaemon(True)
t.start()
while True:
cmd = sys.stdin.readline()
args = cmd.split(' ')
if cmd.find('quit') == 0:
sys.stdout.write('bye-bye\r\n')
break
elif args[0] == 'add_slave':
slave_id = int(args[1])
server.add_slave(slave_id)
sys.stdout.write('done: slave %d added\r\n' % (slave_id))
elif args[0] == 'add_block':
slave_id = int(args[1])
name = args[2]
block_type = int(args[3])
starting_address = int(args[4])
length = int(args[5])
slave = server.get_slave(slave_id)
slave.add_block(name, block_type, starting_address, length)
sys.stdout.write('done: block %s added\r\n' % (name))
elif args[0] == 'set_values':
slave_id = int(args[1])
name = args[2]
address = int(args[3])
values = []
for val in args[4:]:
values.append(int(val))
slave = server.get_slave(slave_id)
slave.set_values(name, address, values)
values = slave.get_values(name, address, len(values))
sys.stdout.write('done: values written: %s\r\n' % (str(values)))
elif args[0] == 'get_values':
slave_id = int(args[1])
name = args[2]
address = int(args[3])
length = int(args[4])
slave = server.get_slave(slave_id)
values = slave.get_values(name, address, length)
sys.stdout.write('done: values read: %s\r\n' % (str(values)))
else:
sys.stdout.write("unknown command %s\r\n" % (args[0]))
finally:
server.stop()
#更新值
def updateData():
global slave_1
while True:
slave_1.set_values('0', 3, 88)
slave_1.set_values('0', 4, 99)
# print(slave_1.get_values('0',3,2))
time.sleep(1)
if __name__ == "__main__":
main()
python 实现 modbus TCP slave
import sys
import modbus_tk
import modbus_tk.defines as cst
from modbus_tk import modbus_tcp
def main():
"""main"""
logger = modbus_tk.utils.create_logger(name="console", record_format="%(message)s")
try:
#Create the server
server = modbus_tcp.TcpServer()
logger.info("running...")
logger.info("enter 'quit' for closing the server")
server.start()
slave_1 = server.add_slave(1)
slave_1.add_block('0', cst.HOLDING_REGISTERS, 0, 100)
while True:
cmd = sys.stdin.readline()
args = cmd.split(' ')
if cmd.find('quit') == 0:
sys.stdout.write('bye-bye\r\n')
break
elif args[0] == 'add_slave':
slave_id = int(args[1])
server.add_slave(slave_id)
sys.stdout.write('done: slave %d added\r\n' % slave_id)
elif args[0] == 'add_block':
slave_id = int(args[1])
name = args[2]
block_type = int(args[3])
starting_address = int(args[4])
length = int(args[5])
slave = server.get_slave(slave_id)
slave.add_block(name, block_type, starting_address, length)
sys.stdout.write('done: block %s added\r\n' % name)
elif args[0] == 'set_values':
slave_id = int(args[1])
name = args[2]
address = int(args[3])
values = []
for val in args[4:]:
values.append(int(val))
slave = server.get_slave(slave_id)
slave.set_values(name, address, values)
values = slave.get_values(name, address, len(values))
sys.stdout.write('done: values written: %s\r\n' % str(values))
elif args[0] == 'get_values':
slave_id = int(args[1])
name = args[2]
address = int(args[3])
length = int(args[4])
slave = server.get_slave(slave_id)
values = slave.get_values(name, address, length)
sys.stdout.write('done: values read: %s\r\n' % str(values))
else:
sys.stdout.write("unknown command %s\r\n" % args[0])
finally:
server.stop()
if __name__ == "__main__":
main()
总结:
这里用python实现了modbus slave,实现modbus master也可以用modbus_tk实现,后续贴出实现modbus master代码。