如何通过pyModbusTCP的DataBank实现Modbus功能码23服务器端写数据？

设备仅支持Modbus功能码23（读写多个寄存器），且仅能作为Modbus客户端和TCP主站。目前可成功从设备读取寄存器2048中的数据，但无法将数据写回至设备的寄存器0中。

实现代码 #

import random
from time import sleep
from pyModbusTCP.server import DataBank, ModbusServer
# instance of DataBank
data_bank = DataBank()

def int_to_bit(value):
    bin_str = format(value, '016b')
    return [int(bit) for bit in bin_str]

def bit_to_int(bin_list):
    bin_str = ''.join(str(bit) for bit in bin_list)
    return int(bin_str, 2)

def print_values(values):
    for i in range(len(values)):
        print(f"Register {i+1}: {values[i]}")

def user_input():
    user_input = input("Enter up to 16 bits (e.g. 1010101010101010): ")
    if not all(bit in '01' for bit in user_input) or len(user_input) > 16:
        print("Invalid input. Please enter a valid 16-bit binary number.")
        return user_input()
    bin_list = [int(bit) for bit in user_input.zfill(16)]
    value = bit_to_int(bin_list)
    data_bank.set_holding_registers(0, [value])
    print(f"{value} written to register 0.")
    print(f"{bin_list} written to register 0.")
        

# instance of ModbusServer
server = ModbusServer('192.168.88.253', 502, no_block=True, data_bank=data_bank)

# Initialize the holding registers to 0 for addresses 0 to 7 and 2048 to 2049
data_bank.set_holding_registers(0, [0]*2)
data_bank.set_holding_registers(2048, [0]*2)

try:
    print("Starting Server...")
    server.start()
    print("Server is online")
    
    previous_state = [0]*2  # to keep track of changes in registers 2048 to 2049
    
    while True:
        # Check for changes in the holding registers at address 2048
        current_state = data_bank.get_holding_registers(2048, 2)
        if current_state != previous_state:
            print("Values at registers 2048 to 2055 have changed:")
            print_values(current_state)
            register_0_value = data_bank.get_holding_registers(2048, 1)[0]
            bin_list = int_to_bit(register_0_value)
            print(f"Register 0 value in 16-bits: {bin_list}")
            previous_state = current_state  # update previous state

        user_input()

        sleep(2)
except Exception as e:
    print(f"An error occurred: {e}")

print("Shutdown server...")
server.stop()
print("Server is stopped")

已尝试的解决方法 #

• 在user_input()函数中替换set_holding_registers()为其他方法
• 尝试传入二进制列表（如[0,1,1,1,0...]）而非整数值
• 尝试使用库中server.py的server._write_read_multiple_registers()方法，但未能掌握正确用法

内容的提问来源于stack exchange，提问作者Shivansh Sharma