brandon_l
使用python解決工業通信問題是一個非常好的選擇，python具有豐富的生態，可以輕鬆解決工業通信的各種問題。 本篇主要介紹使用pymodbus庫進行modbus tcp模擬，實現pc端讀取plc或工業設備modbus變數。
安裝pymodbus：
pip install -U pymodbus
01. 建立Modbus TCP Server
這裡我們先創建一個虛擬的modbus設備，如果你手裡有一個plc或者工業設備，可以直接跳過本節。
檔案： modbus_server.py
from pymodbus.server.sync import (
StartTcpServer,
)
from pymodbus.datastore import (
ModbusSequentialDataBlock,
ModbusServerContext,
ModbusSlaveContext,
)
from pymodbus.version import version
datablock = ModbusSequentialDataBlock.create()
context = ModbusSlaveContext(
di=datablock,
co=datablock,
hr=datablock,
ir=datablock,
)
single = True
# Build data storage
store = ModbusServerContext(slaves=context, single=single)
if __name__ == '__main__':
address = ("0.0.0.0", 503)
StartTcpServer(
context=store, # Data storage
address=address, # listen address
allow_reuse_address=True, # allow the reuse of an address
)
直接運行該腳本，就可以在本機的503埠創建一台modbus設備了，具體實現暫不深追，我們學習的重點是用戶端對modbus變數的讀寫。
02. 讀寫modbus變數
modbus變數類型以及位址
| 物件類型 | 訪問 | 大小 | 位址 |
|---|---|---|---|
| 線圈 | 讀寫 | 1 位 | 00001 – 09999 |
| 離散輸入 | 唯讀 | 1 位 | 10001 – 19999 |
| 輸入寄存器 | 唯讀 | 16 位 | 30001 – 39999 |
| 持有登記冊 | 讀寫 | 16 位 | 40001 – 49999 |
coil是線圈，Discrete input是數位量輸入，Input register是類比量輸入，Holding register是保持寄存器。一般位址範圍是0-65535
2.1 讀取常規變數
讀寫線圈 | 讀取輸入變數 | 讀寫保持寄存器
from pymodbus.client.sync import ModbusTcpClient
from pymodbus.bit_read_message import ReadCoilsResponse
from pymodbus.register_read_message import ReadInputRegistersResponse
from pymodbus.exceptions import ConnectionException # 连接失败，用于异常处理
host = '127.0.0.1'
port = 503
client = ModbusTcpClient(host,port)
# 写入线圈
client.write_coil(1, True)
client.write_coil(2, False)
client.write_coil(3, True)
# 读取线圈 注意对于离散量的读取，第二个参数cout是有坑的，必须为8的倍数个
result:ReadCoilsResponse = client.read_coils(address=1,cout=8) # 从地址1开始读，读取8个线圈，一次读8的倍数个线圈，不设置为8的倍数可能会出现问题
print(result.isError())
# 不建议使用
print(result.getBit(7)) # 这里的参数address不是plc里的地址，而是python列表的address，
print('read_coils ')
# 建议使用
print(result.bits) # 打印读取结果，一共8位
# 读取其中的位
print(
result.bits[0],
result.bits[1],
result.bits[2]
) # 相当于result.getBit(0)
# 读取数字输入
result = client.read_discrete_inputs(address=10001,count=8) # 从10001开始读，读取8位
print(result.bits)
# 读取模拟输入寄存器
input_register_result:ReadInputRegistersResponse = client.read_input_registers(1,count=8)
# print(f'is_error:{input_register_result.isError()}')
print('read_input_registers ')
print(input_register_result.registers)
print(input_register_result.getRegister(0))
# 读写保持寄存器
client.write_register(address=40001,value=100)
result:ReadInputRegistersResponse = client.read_holding_registers(address=40001,count=1)
print('read_holding_registers ')
print(result.registers)
＃ 关闭连接
client.close()
2.2 讀取複雜變數
字串、浮點數、負數等。 這裡需要注意modbus設備的存儲結構是低位低位元組還是低位高位元組，也就是設備記憶體的位元組、字的排列順序。
| 詞序 | 位元組順序 | 字1 | 字2 |
|---|---|---|---|
| 大 | 大 | 0x1234 | 0x5678 |
| 大 | 小 | 0x3412 | 0x7856 |
| 小 | 大 | 0x5678 | 0x1234 |
| 小 | 小 | 0x7856 | 0x3412 |
檔案：
# 复杂数据类型
from collections import OrderedDict
import logging
from pymodbus.client.sync import ModbusTcpClient as ModbusClient
from pymodbus.constants import Endian
from pymodbus.payload import BinaryPayloadBuilder, BinaryPayloadDecoder
ORDER_DICT = {"<": "LITTLE", ">": "BIG"}
def run_binary_payload_client(host:str,port:int):
for word_endian, byte_endian in (
(Endian.Big, Endian.Big),
(Endian.Big, Endian.Little),
(Endian.Little, Endian.Big),
(Endian.Little, Endian.Little),
):
print("-" * 60)
print(f"Word Order: {ORDER_DICT[word_endian]}")
print(f"Byte Order: {ORDER_DICT[byte_endian]}")
print()
builder = BinaryPayloadBuilder(
wordorder=word_endian,
byteorder=byte_endian,
)
# 写入的变量
my_string = "abcd-efgh123345765432"
builder.add_string(my_string)
builder.add_bits([0, 1, 0, 1, 1, 0, 1, 0])
builder.add_8bit_int(-0x12)
builder.add_8bit_uint(0x12)
builder.add_16bit_int(-0x5678)
builder.add_16bit_uint(0x1234)
builder.add_32bit_int(-0x1234)
builder.add_32bit_uint(0x12345678)
builder.add_16bit_float(12.34)
builder.add_16bit_float(-12.34)
builder.add_32bit_float(22.34)
builder.add_32bit_float(-22.34)
builder.add_64bit_int(-0xDEADBEEF)
builder.add_64bit_uint(0x12345678DEADBEEF)
builder.add_64bit_uint(0x12345678DEADBEEF)
builder.add_64bit_float(123.45)
builder.add_64bit_float(-123.45)
registers = builder.to_registers()
print("Writing Registers:")
print(registers)
print("\n")
payload = builder.build()
address = 40001 # 从40001开始写入
# We can write registers
client.write_registers(address, registers, unit=1) # 写入
# 读取复杂变量
print("Reading Registers:")
address = 40001
count = len(payload)
print(f"payload_len {count}")
result = client.read_holding_registers(address, count, slave=1)
print(result.registers)
print("\n")
decoder = BinaryPayloadDecoder.fromRegisters(
result.registers, byteorder=byte_endian, wordorder=word_endian
)
# Make sure word/byte order is consistent between BinaryPayloadBuilder and BinaryPayloadDecoder
assert (
decoder._byteorder == builder._byteorder # pylint: disable=protected-access
) # nosec
assert (
decoder._wordorder == builder._wordorder # pylint: disable=protected-access
) # nosec
decoded = OrderedDict(
[
("string", decoder.decode_string(len(my_string))),
("bits", decoder.decode_bits()),
("8int", decoder.decode_8bit_int()),
("8uint", decoder.decode_8bit_uint()),
("16int", decoder.decode_16bit_int()),
("16uint", decoder.decode_16bit_uint()),
("32int", decoder.decode_32bit_int()),
("32uint", decoder.decode_32bit_uint()),
("16float", decoder.decode_16bit_float()),
("16float2", decoder.decode_16bit_float()),
("32float", decoder.decode_32bit_float()),
("32float2", decoder.decode_32bit_float()),
("64int", decoder.decode_64bit_int()),
("64uint", decoder.decode_64bit_uint()),
("ignore", decoder.skip_bytes(8)),
("64float", decoder.decode_64bit_float()),
("64float2", decoder.decode_64bit_float()),
]
)
print("Decoded Data")
for name, value in iter(decoded.items()):
print(
"%s\t" % name, # pylint: disable=consider-using-f-string
hex(value) if isinstance(value, int) else value,
)
print("\n")
# 关闭连接
client.close()
if __name__ == "__main__":
run_binary_payload_client("127.0.0.1", 503)
相關資料
1、Modbus TCP 通信協定
2、Modbus TCP 通信協定2
3、PyModbus官方文檔
4、PyModBus倉庫
Homin's Library 