快捷導(dǎo)航

使用pymodbus模塊實(shí)現(xiàn)Modbus通訊方式

更新時(shí)間：2026年06月02日 08:42:57 作者：草莓仙生

這段文章詳細(xì)介紹了如何使用PyModbus庫進(jìn)行Modbus通信,涵蓋建立通訊、讀取與寫入寄存器、處理異常及數(shù)據(jù)類型轉(zhuǎn)換等內(nèi)容,適合Python開發(fā)者參考

1. 建立通訊

from pymodbus.client import ModbusTcpClient

client = ModbusTcpClient('localhost', port=502)
client.connect()

2. 讀取數(shù)據(jù)

2.1 讀取寄存器

Modbus協(xié)議定義了幾種類型的寄存器，最常見的是保持寄存器和輸入寄存器。

以下示例展示了如何讀取保持寄存器：

response = client.read_holding_registers(address=1, count=10, unit=1)
if not response.isError():
    print("Register Values: ", response.registers)
else:
    print("Failed to read registers")

在這個(gè)例子中，read_holding_registers方法用于讀取地址為1的起始位置、數(shù)量為10的連續(xù)寄存器。unit參數(shù)表示從哪個(gè)單元（即設(shè)備ID）讀取數(shù)據(jù)。

注意：pymodbus某個(gè)版本已將unit字段改為slave。使用時(shí)即使寫錯(cuò)也不會(huì)報(bào)錯(cuò)，注意查看你的pymodbus文檔。

3. 寫入數(shù)據(jù)

3.1 寫入單個(gè)寄存器

要向設(shè)備的單個(gè)寄存器寫入數(shù)據(jù)，可以使用以下代碼：

write_response = client.write_register(address=1, value=25, unit=1)
if not write_response.isError():
    print("Written successfully")
else:
    print("Failed to write register")

這里使用了write_register方法，它接受地址、要寫入的值以及目標(biāo)設(shè)備的單元ID。

3.2 寫入多個(gè)寄存器

如果要寫入多個(gè)寄存器，可以使用write_registers方法：

values = [20, 40, 60, 80, 100]
write_response = client.write_registers(address=1, values=values, unit=1)
if not write_response.isError():
    print("Multiple registers written successfully")
else:
    print("Failed to write multiple registers")

這里values列表包含了要寫入寄存器的值序列。

4. 處理異常

處理Modbus通信過程中可能出現(xiàn)的異常非常重要，可以使用try-except語句捕獲這些異常：

from pymodbus.exceptions import ModbusException

try:
    # 嘗試執(zhí)行Modbus操作
    response = client.read_holding_registers(address=1, count=10, unit=1)
except ModbusException as ex:
    print("An error occurred:", str(ex))

5. 數(shù)據(jù)類型轉(zhuǎn)換

modbus讀寫數(shù)據(jù)時(shí)的數(shù)據(jù)類型轉(zhuǎn)換通常是通過struct模塊實(shí)現(xiàn)的。

讀取modbus數(shù)據(jù)時(shí)，常用到以下的數(shù)據(jù)類型轉(zhuǎn)換關(guān)系。

 def _convert_to_float(self, registers):
        if len(registers) != 2:
            print("Invalid register length for float")
            return None
        combined = (registers[0] << 16) + registers[1]
        return struct.unpack(">f", combined.to_bytes(4, byteorder="big"))[0]

    def _convert_to_int32(self, registers):
        if len(registers) != 2:
            print("Invalid register length for int32")
            return None
        combined = (registers[0] << 16) + registers[1]
        return struct.unpack(">i", combined.to_bytes(4, byteorder="big"))[0]

    def _convert_to_uint32(self, registers):
        if len(registers) != 2:
            print("Invalid register length for uint32")
            return None
        combined = (registers[0] << 16) + registers[1]
        return struct.unpack(">I", combined.to_bytes(4, byteorder="big"))[0]

    def _convert_to_int16(self, registers):
        if len(registers) != 1:
            print("Invalid register length for int16")
            return None
        return struct.unpack(">h", struct.pack(">H", registers[0]))[0]

    def _convert_to_uint16(self, registers):
        if len(registers) != 1:
            print("Invalid register length for uint16")
            return None
        return struct.unpack(">H", struct.pack(">H", registers[0]))[0]

寫入modbus數(shù)據(jù)時(shí)，常用到以下的數(shù)據(jù)類型轉(zhuǎn)換關(guān)系。

 def _int32_to_registers(self, value):
        packed = struct.pack(">i", value)
        high_register, low_register = struct.unpack(">HH", packed)
        return [high_register, low_register]

    def _uint32_to_registers(self, value):
        packed = struct.pack(">I", value)
        high_register, low_register = struct.unpack(">HH", packed)
        return [high_register, low_register]

    def _uint16_to_register(self, value):
        packed = struct.pack(">H", value)
        register = struct.unpack(">H", packed)[0]
        return [register]

    def _int16_to_register(self, value):
        packed = struct.pack(">h", value)
        register = struct.unpack(">h", packed)[0]
        return [register]