''' READ_COILS = 0x01 读线圈寄存器 READ_DISCRETE_INPUTS = 0x02 读离散输入寄存器 READ_HOLDING_REGISTERS = 0x03 读保持寄存器 READ_INPUT_REGISTER = 0x04 读输入寄存器 H05 = 0x05 写单个线圈寄存器 WRITE_MULTIPLE_COILS = 0x0F 写多个线圈寄存器 WRITE_SINGLE_REGISTER = 0x06 写单个保持寄存器 WRITE_MULTIPLE_REGISTERS = 0x10 写多个保持寄存器 ''' from serial import Serial import machine import time class modbus_host: def __init__(self, port=2, baudrate = 9600): self.modbus_485 = Serial(port, baudrate) ######################### READ_COILS ######################### ''' slave_addr 设备地址 starting_address 寄存器开始地址 coil_quantity 读取数据个数 最少8个 ''' def H01_read_coils(self, slave_addr, starting_address, coil_quantity): coil_status = self.modbus_485.read_coils(slave_addr, starting_address, coil_quantity) print('Coil status: ' + ' '.join('{:d}'.format(x) for x in coil_status)) return coil_status # ##################### READ_DISCRETE_INPUTS ################## ''' slave_addr 设备地址 starting_address 寄存器开始地址 input_quantity 读取数据个数 ''' def H02_read_discrete_input(self, slave_addr, starting_address, input_quantity): input_status = self.modbus_485.read_discrete_inputs(slave_addr, starting_address, input_quantity) print('Input status: ' + ' '.join('{:d}'.format(x) for x in input_status)) return input_status # ##################### READ HOLDING REGISTERS ################## ''' slave_addr 设备地址 starting_address 寄存器开始地址 register_quantity 读取数据个数 signed 是否为有符号数 ''' def H03_read_hold_register(self, slave_addr, starting_address, register_quantity, signed=True): register_value = self.modbus_485.read_holding_registers(slave_addr, starting_address, register_quantity, signed) # print('Holding register value: ' + ' '.join('{:d}'.format(x) for x in register_value)) return register_value # ##################### READ INPUT REGISTERS ################## ''' slave_addr 设备地址 starting_address 寄存器开始地址 register_quantity 读取数据个数 signed 是否为有符号数 ''' def H04_read_input_register(self, slave_addr, starting_address, register_quantity, signed=True): register_value = self.modbus_485.read_input_registers(slave_addr, starting_address, register_quantity, signed) #print(f'Input register value:{register_value}') #print('Input register value: ' + ' '.join('{:d}'.format(x) for x in register_value)) return register_value ##################### WRITE SINGLE COIL ################## ''' slave_addr 设备地址 output_address 寄存器地址 output_value 写入数据 ''' def H05_write_coil(self, slave_addr, output_address, output_value): return self.modbus_485.write_single_coil(slave_addr, output_address, output_value) ##################### WRITE SINGLE REGISTER ################## ''' slave_addr 设备地址 register_address 寄存器地址 register_value 写入数据 signed 是否为有符号数 ''' def H06_write_single_register(self, slave_addr, register_address, register_value, signed=True): return self.modbus_485.write_single_register(slave_addr, register_address, register_value, signed) # ##################### WRITE MULIPLE COILS ################## ''' slave_addr 设备地址 starting_address 寄存器地址 output_values 写入数据 ''' def H0F_write_multi_coils(self, slave_addr, starting_address, output_values): return self.modbus_485.write_multiple_coils(slave_addr, starting_address, output_values) # ##################### WRITE MULIPLE REGISTERS ################## ''' slave_addr 设备地址 register_address 寄存器地址 register_values 写入数据 signed 是否为有符号数 ''' def H10_write_multi_registers(self, slave_addr, register_address, register_values, signed=True): return self.modbus_485.write_multiple_registers(slave_addr, register_address, register_values, signed) if __name__ == '__main__': ''' control_coil() ##依次控制DO1-DO4 H0f(2,0,[1,1,1,1]) #开启全部DO time.sleep(1) # H0f(2,0,[0,0,0,0]) #关闭全部DO H02(2,0,4) #获取4路DI数值 write_Registers() #控制采样寄存器 ''' modbus_485 = modbus_host(2, 9600) modbus_485.H04_read_input_register(2, 0, 4) #获取温度传感器 addr 02 4路AI数值 温度 modbus_485.H03_read_hold_register(1, 400, 12) #获取12路AI数值 addr 01 12路流量 输出点 控制流量计 modbus_485.H10_write_multi_registers(1, 400, [1000, 100, 200, 300, 400, 500, 600, 500, 500, 500, 0, 0]) modbus_485.H03_read_hold_register(1, 400, 12) #获取12路AI数值 addr 01 12路流量 modbus_485.H04_read_input_register(1, 0, 12) #获取12路AI数值 输入点 流量计输入