|
|
|
|
|
import serial
|
|
|
|
|
|
import modbus_tk
|
|
|
import modbus_tk.defines as cst
|
|
|
from modbus_tk import modbus_rtu
|
|
|
import sys
|
|
|
|
|
|
from .ByteOrder import *
|
|
|
from .SetMessage import *
|
|
|
|
|
|
# master = modbus_rtu.RtuMaster(serial.Serial(port = 'COM4', baudrate = 9600, bytesize = 8, parity = 'N', stopbits = 1, xonxoff=0))
|
|
|
|
|
|
# exec(slave=1, function_code=READ_HOLDING_REGISTERS, starting_address=0, quantity_of_x=0, output_value=0, data_format=“”, expected_length=-1)
|
|
|
# 参数说明：
|
|
|
# @slave=1 : identifier of the slave. from 1 to 247.
|
|
|
# @function_code=READ_HOLDING_REGISTERS：功能码
|
|
|
# @starting_address=100：寄存器的开始地址
|
|
|
# @quantity_of_x=3：寄存器/线圈的数量
|
|
|
# @output_value：一个整数或可迭代的值：1/[1,1,1,0,0,1]/xrange(12)
|
|
|
# @data_format：对接收的数据进行格式化
|
|
|
|
|
|
|
|
|
|
|
|
class RTUMaster():
|
|
|
def __init__(self, port = 'COM4', baudrate = 9600, bytesize = 8, parity = 'N', stopbits = 1, xonxoff = 0):
|
|
|
try:
|
|
|
self.master = modbus_rtu.RtuMaster(serial.Serial(port = port, baudrate = baudrate, bytesize = bytesize, parity = parity, stopbits = stopbits, xonxoff = xonxoff))
|
|
|
self.master.set_timeout(5.0)
|
|
|
self.master.set_verbose(True)
|
|
|
except Exception as e:
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
|
|
def writeSingleRegister(self, slaveId, address, outputValue, order='ABCD'):
|
|
|
try:
|
|
|
if '.' not in str(outputValue):
|
|
|
# 整数值，使用单寄存器写入
|
|
|
outputValue = int(outputValue)
|
|
|
self.master.execute(slaveId, cst.WRITE_SINGLE_REGISTER, starting_address=address, output_value=outputValue)
|
|
|
else:
|
|
|
# 浮点数值，需要使用多寄存器写入（因为浮点数占用2个寄存器）
|
|
|
outputValue = float(outputValue)
|
|
|
valueByte = None # 初始化变量
|
|
|
|
|
|
if order == 'ABCD': # 大端模式
|
|
|
valueByte = floatToABCD(outputValue)
|
|
|
elif order == 'DCBA': # 小端模式
|
|
|
valueByte = floatToDCBA(outputValue)
|
|
|
elif order == 'BADC':
|
|
|
valueByte = floatToBADC(outputValue)
|
|
|
elif order == 'CDAB':
|
|
|
valueByte = floatToCDAB(outputValue)
|
|
|
else:
|
|
|
# 默认使用 ABCD 字节序
|
|
|
# print(f"Unknown byte order '{order}', using default ABCD")
|
|
|
valueByte = floatToABCD(outputValue)
|
|
|
|
|
|
if valueByte is not None:
|
|
|
# 浮点数必须使用 WRITE_MULTIPLE_REGISTERS，因为需要写入2个寄存器
|
|
|
self.master.execute(slaveId, cst.WRITE_MULTIPLE_REGISTERS, starting_address=address, output_value=valueByte)
|
|
|
else:
|
|
|
# print(f"Failed to convert float value {outputValue} with order {order}")
|
|
|
return 'error'
|
|
|
except Exception as e:
|
|
|
print(f'writeSingleRegister error: {e}')
|
|
|
return 'error'
|
|
|
|
|
|
def writeSingleCoil(self, slaveId, address, outputValue):
|
|
|
try:
|
|
|
outputValue = int(outputValue)
|
|
|
if outputValue in [0, 1]:
|
|
|
self.master.execute(slaveId, cst.WRITE_SINGLE_COIL, address, output_value=outputValue)
|
|
|
except Exception as e:
|
|
|
return 'error'
|
|
|
|
|
|
def writeMultipleRegisters(self, slaveId, startAddress, outputValues, order='ABCD'):
|
|
|
"""写多个寄存器，支持不同数据类型和字节序"""
|
|
|
try:
|
|
|
processedValues = []
|
|
|
for value in outputValues:
|
|
|
if '.' not in str(value):
|
|
|
# 整数值
|
|
|
processedValues.append(int(value))
|
|
|
else:
|
|
|
# 浮点值 - 根据字节序转换为寄存器对
|
|
|
floatValue = float(value)
|
|
|
if order == 'ABCD': # 大端模式
|
|
|
valueByte = floatToABCD(floatValue)
|
|
|
elif order == 'DCBA': # 小端模式
|
|
|
valueByte = floatToDCBA(floatValue)
|
|
|
elif order == 'BADC':
|
|
|
valueByte = floatToBADC(floatValue)
|
|
|
elif order == 'CDAB':
|
|
|
valueByte = floatToCDAB(floatValue)
|
|
|
processedValues.extend(valueByte)
|
|
|
|
|
|
self.master.execute(slaveId, cst.WRITE_MULTIPLE_REGISTERS, starting_address=startAddress, output_value=processedValues)
|
|
|
except Exception as e:
|
|
|
return 'error'
|
|
|
|
|
|
def writeMultipleCoils(self, slaveId, startAddress, outputValues):
|
|
|
"""写多个线圈，支持布尔值列表或0/1整数列表"""
|
|
|
try:
|
|
|
processedValues = []
|
|
|
for value in outputValues:
|
|
|
if isinstance(value, bool):
|
|
|
processedValues.append(1 if value else 0)
|
|
|
else:
|
|
|
intValue = int(value)
|
|
|
if intValue in [0, 1]:
|
|
|
processedValues.append(intValue)
|
|
|
else:
|
|
|
# 如果值不是0或1，转换为布尔值
|
|
|
processedValues.append(1 if intValue else 0)
|
|
|
|
|
|
self.master.execute(slaveId, cst.WRITE_MULTIPLE_COILS, starting_address=startAddress, output_value=processedValues)
|
|
|
except Exception as e:
|
|
|
return 'error'
|
|
|
|
|
|
|
|
|
def readHoldingRegisters(self, slaveId, startAddress, varNums, order = 'ABCD'):
|
|
|
try:
|
|
|
if order == 'int':
|
|
|
valueByte = self.master.execute(slaveId, cst.READ_HOLDING_REGISTERS, startAddress, varNums)[0]
|
|
|
else:
|
|
|
value = self.master.execute(slaveId, cst.READ_HOLDING_REGISTERS, startAddress, 2)
|
|
|
if order == 'ABCD': # 大端模式
|
|
|
valueByte = ABCDToFloat(value)
|
|
|
elif order == 'DCBA': # 小端模式
|
|
|
valueByte = DCBAToFloat(value)
|
|
|
elif order == 'BADC':
|
|
|
valueByte = BADCToFloat(value)
|
|
|
elif order == 'CDAB':
|
|
|
valueByte = CDABToFloat(value)
|
|
|
return valueByte
|
|
|
except Exception as e:
|
|
|
return 'error'
|
|
|
|
|
|
def readInputRegisters(self, slaveId, startAddress, varNums, order = 'ABCD'):
|
|
|
try:
|
|
|
if order == 'int':
|
|
|
valueByte = self.master.execute(slaveId, cst.READ_INPUT_REGISTERS, startAddress, varNums)[0]
|
|
|
else:
|
|
|
value = self.master.execute(slaveId, cst.READ_INPUT_REGISTERS, startAddress, 2)
|
|
|
if order == 'ABCD': # 大端模式
|
|
|
valueByte = ABCDToFloat(value)
|
|
|
elif order == 'DCBA': # 小端模式
|
|
|
valueByte = DCBAToFloat(value)
|
|
|
elif order == 'BADC':
|
|
|
valueByte = BADCToFloat(value)
|
|
|
elif order == 'CDAB':
|
|
|
valueByte = CDABToFloat(value)
|
|
|
return valueByte
|
|
|
except Exception as e:
|
|
|
return 'error'
|
|
|
|
|
|
def readCoils(self, slaveId, startAddress, varNums, order='ABCD'):
|
|
|
try:
|
|
|
value = self.master.execute(slaveId, cst.READ_COILS, startAddress, varNums)
|
|
|
# 如果只读取一个线圈，返回单个值；否则返回列表
|
|
|
return value[0]
|
|
|
except Exception as e:
|
|
|
return 'error'
|
|
|
|
|
|
def readInputCoils(self, slaveId, startAddress, varNums):
|
|
|
try:
|
|
|
value = self.master.execute(slaveId, cst.READ_DISCRETE_INPUTS, startAddress, varNums)
|
|
|
# 如果只读取一个线圈，返回单个值；否则返回列表
|
|
|
return value[0]
|
|
|
except Exception as e:
|
|
|
return 'error'
|
|
|
|
|
|
# -*- coding: utf_8 -*-
|
|
|
|
|
|
|
|
|
# import serial
|
|
|
# import modbus_tk
|
|
|
# import modbus_tk.defines as cst
|
|
|
# from modbus_tk import modbus_rtu
|
|
|
|
|
|
|
|
|
# def mod(PORT="com2"):
|
|
|
# red = []
|
|
|
# alarm = ""
|
|
|
# try:
|
|
|
# # 设定串口为从站
|
|
|
# master = modbus_rtu.RtuMaster(serial.Serial(port=PORT,
|
|
|
# baudrate=9600, bytesize=8, parity='N', stopbits=1))
|
|
|
# master.set_timeout(5.0)
|
|
|
# master.set_verbose(True)
|
|
|
|
|
|
# # 读保持寄存器
|
|
|
# red = master.execute(1, cst.READ_HOLDING_REGISTERS, 0, 9) # 这里可以修改需要读取的功能码
|
|
|
# print(red)
|
|
|
# alarm = "正常"
|
|
|
# return list(red), alarm#
|
|
|
|
|
|
# except Exception as exc:
|
|
|
# print(str(exc))
|
|
|
# alarm = (str(exc))
|
|
|
|
|
|
# return red, alarm ##如果异常就返回[],故障信息
|
|
|
|
|
|
|
|
|
# if __name__ == "__main__":
|
|
|
# mod()
|