Pythonpython_serial

python 玩转 modbus

2018-11-09  本文已影响25人  时尚灬IT男

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代码。

上一篇下一篇

猜你喜欢

热点阅读