Python pymodbus类库使用学习总结

李玲玲 · 发表于 2024-8-4 20:02:17

实践环境

Python 3.9.13
https://www.python.org/ftp/python/3.9.13/python-3.9.13-amd64.exe
pymodbus-3.6.8-py3-none-any.whl
https://files.pythonhosted.org/packages/35/19/a9d16f74548d6750acf6604fa74c2cd165b5bc955fe021bf5e1fa04acf14/pymodbus-3.6.8-py3-none-any.whl
pyserial-3.5
备注：如果不安装该模块，采用串口通信时（下述代码中 comm = 'serial'时的通信），会报错NameError: name 'serial' is not defined

pip install pyserial

复制代码

Virtual Serial Port Driver 6.9
链接：https://pan.baidu.com/s/15pCeYyOZWSsbEr_rU1e5hg?pwd=vspd
提取码：vspd
代码实践

修改pymodbus\logging.py

路径： PYTHON_HOME\Lib\site-packages\pymodbus\logging.py
修改源码是为了更方便使用类库自带日志打印器

# ...略
class Log:
"""Class to hide logging complexity.
:meta private:
"""
# _logger = logging.getLogger(__name__) # commented by shouke
_logger = logging.getLogger('logger') # added by shouke # 方便在不同模块获取日志打印器
@classmethod
def apply_logging_config(cls, level, log_file_name):
"""Apply basic logging configuration."""
if level == logging.NOTSET:
level = cls._logger.getEffectiveLevel()
if isinstance(level, str):
level = level.upper()
log_stream_handler = logging.StreamHandler()
log_formatter = logging.Formatter(
"%(asctime)s %(levelname)-5s %(module)s:%(lineno)s %(message)s"
)
log_stream_handler.setFormatter(log_formatter)
##################### commented by shouke #################
# cls._logger.addHandler(log_stream_handler)
# if log_file_name:
# log_file_handler = logging.FileHandler(log_file_name)
# log_file_handler.setFormatter(log_formatter)
# cls._logger.addHandler(log_file_handler)
##########################################################
##################### added by shouke #################
if not cls._logger.handlers: # 如果不加这个判断，在不同模块都执行pymodbus_apply_logging_config函数时，会导致同一条日志被重复打印
cls._logger.addHandler(log_stream_handler)
if log_file_name:
log_file_handler = logging.FileHandler(log_file_name)
log_file_handler.setFormatter(log_formatter)
cls._logger.addHandler(log_file_handler)
##########################################################
cls.setLevel(level)
# ...略
pymodbus_apply_logging_config(level=logging.INFO) # 统一日志打印器配置

复制代码

异步服务器和异步客户端实现

异步服务器代码实现

server_async.py

#!/usr/bin/env python3
'''Pymodbus 异步服务器示例
多线程异步服务器的一个示例。
'''
import asyncio
import logging
### 不修改 PYTHON_HOME\Lib\site-packages\pymodbus\logging.py源码的基础上，获取日志打印器相关方法及说明：
# 方法1
# _logger = logging.getLogger(__name__) # 采用该_logger打印的日志看不到
# logging.basicConfig(level=logging.DEBUG) # 不注释上行代码的基础上添加这行代码，日志能打印出来，但是日志所在模块不为当前模块
# 方法2
# _logger = logging.getLogger(__file__)
# logging.basicConfig(level=logging.INFO) # 添加这行代码，确保日志能打印出来，但是打印出来的日志所在模块不为当前模块
# 方法3：
# from pymodbus.logging import Log,pymodbus_apply_logging_config
# pymodbus_apply_logging_config(level=logging.INFO) # 如果缺少这行代码，下面async_helper中的日志打印将无法在控制台输出
# _logger = Log._logger
# 修改 PYTHON_HOME\Lib\site-packages\pymodbus\logging.py源码的基础上，获取日志打印器相关方法：
_logger = logging.getLogger('logger')
# _logger.setLevel(logging.INFO)
from pymodbus import __version__ as pymodbus_version
from pymodbus.datastore import (
ModbusSequentialDataBlock,
ModbusServerContext,
ModbusSlaveContext,
ModbusSparseDataBlock,
)
from pymodbus.device import ModbusDeviceIdentification
from pymodbus.server import (
StartAsyncSerialServer,
StartAsyncTcpServer,
StartAsyncTlsServer,
StartAsyncUdpServer,
)
# hook函数
def server_request_tracer(request, *_addr):
"""跟踪请求
所有服务器请求在被处理之前都经过此过滤器
"""
_logger.info(f'---> REQUEST: {request}') # 输出，类似这样： ---> REQUEST: ReadBitRequest(0,1)
def server_response_manipulator( response):
"""操纵响应
所有服务器响应在发送之前都通过此过滤器
过滤器返回：
- 响应，原始或修改后的
- 跳过编码，发出是否对响应进行编码的信号
"""
_logger.info(f'---> RESPONSE: {response}') # 输出，类似这样：---> RESPONSE: ReadCoilsResponse(1)
# response.should_respond = False # 如果让该行代码生效，则客户端收不到服务器响应
return response, False
class Args:
comm = 'tcp' # 通讯模式，可选值 tcp、udp serial、tls
comm_defaults = {
'tcp': ['socket', 5020],
# 'tcp': ['rtu', 5020], # 如果采用ModbusRTU协议则使用这个
'udp': ['socket', 5020],
# 'serial': ['rtu', '/dev/ptyp0'], # Linux
'serial': ['rtu', 'COM1'] # Windows(本例中COM1和COM2是成对的)
'tls': ['tls', 5020]
}
framer = comm_defaults[comm][0] # 帧类型
port = comm_defaults[comm][1] # 站点端口
host = ''
baudrate = 9600 # 串行设备波特率即每秒传输的位数
# 连续的、无间隙顺序存储数据块（寄存器块）
datablock = lambda : ModbusSequentialDataBlock(0x01, [17] * 100) # pylint: disable=unnecessary-lambda-assignment
# 连续的,或者可能有间隙的稀疏、不规则存储数据块
# datablock = lambda : ModbusSparseDataBlock({0x00: 0, 0x05: 1}) # pylint: disable=unnecessary-lambda-assignment
# 工厂模式
# datablock = lambda : ModbusSequentialDataBlock.create() # pylint: disable=unnecessary-lambda-assignment,unnecessary-lambda
# 如果从节点数量不为0
#服务器使用服务器上下文，该上下文允许服务器，以针对不同的从设备ID使用不同的从站上下文(slave context)进行响应。
# 默认情况下，它将为提供的每个slave ID返回相同的上下文（广播模式）。
# 但是，可以通过将single标识设置为False并且提供slave ID到上下文映射的字典来覆盖这种行为：
# 从机上下文也可以按zero_mode初始化，这意味着到地址（0-7）的请求将映射到地址（0-7）。
# 默认值为False，因此地址（0-7）将映射到（1-8）：
# context = {
# 0x01: ModbusSlaveContext( # 0x01为从设备、从机地址
# di=datablock(), # 输入离散量
# co=datablock(), # 输出线圈
# hr=datablock(), # 保持寄存器
# ir=datablock(), # 输入寄存器
# ),
# 0x02: ModbusSlaveContext(
# di=datablock(),
# co=datablock(),
# hr=datablock(),
# ir=datablock(),
# ),
# 0x03: ModbusSlaveContext(
# di=datablock(),
# co=datablock(),
# hr=datablock(),
# ir=datablock(),
# zero_mode=True,
# ),
# }
# single = False
#如果从节点数量为0
context = ModbusSlaveContext(
di=datablock(), co=datablock(), hr=datablock(), ir=datablock()
)
single = True
# 构建数据存储
context = ModbusServerContext(slaves=context,
single=single
)
# ----------------------------------------------------------------------- #
# 初始化服务器信息
# 不对属性字段做任何设置，则字段值默认为空字符串
# ----------------------------------------------------------------------- #
identity = ModbusDeviceIdentification(
info_name={
"VendorName": "Pymodbus",
"ProductCode": "PM",
"VendorUrl": "https://github.com/pymodbus-dev/pymodbus/",
"ProductName": "Pymodbus Server",
"ModelName": "Pymodbus Server",
"MajorMinorRevision": pymodbus_version,
}
)
# sslctx=sslctx, # 用于TLS的SSLContext (默认为None, 自动创建)
cert_file_path = './certificates/pymodbus.crt' # 用于TLS 证书文件路径 (如果sslctx为None时使用该文件)
key_file_path = './certificates/pymodbus.key' # 用于TLS 私钥文件路径 (如果sslctx为None时使用该文件)
async def run_async_server():
"""Run server."""
txt = f'### start ASYNC server, listening on {Args.port} - {Args.comm}'
_logger.info(txt)
if Args.comm == 'tcp':
address = (Args.host if Args.host else '', Args.port if Args.port else None)
server = await StartAsyncTcpServer(
context=Args.context, # 数据存储
identity=Args.identity, # 服务器标识
# TBD host=
# TBD port=
address=address, # 监听地址
# custom_functions=[], # 允许自定义处理函数
framer=Args.framer, # 使用的帧策略
# ignore_missing_slaves=True, # 忽略对缺失的slave的请求
# broadcast_enable=False, # 是否允许广播将 slave id 0视为广播地址
# timeout=1 # 等待请求完成的时间 # waiting time for request to complete
# 可选参数，实现hook功能
request_tracer=server_request_tracer,
response_manipulator=server_response_manipulator
)
elif Args.comm == 'udp':
address = (
Args.host if Args.host else "127.0.0.1",
Args.port if Args.port else None,
)
server = await StartAsyncUdpServer(
context=Args.context,
identity=Args.identity,
address=address,
# custom_functions=[],
framer=Args.framer,
# ignore_missing_slaves=True,
# broadcast_enable=False,
# timeout=1
)
elif Args.comm == 'serial':
# socat -d -d PTY,link=/tmp/ptyp0,raw,echo=0,ispeed=9600
# PTY,link=/tmp/ttyp0,raw,echo=0,ospeed=9600
server = await StartAsyncSerialServer(
context=Args.context,
identity=Args.identity,
# timeout=1, # waiting time for request to complete
port=Args.port, # 串口
# custom_functions=[],
framer=Args.framer,
# stopbits=1, # 要使用的停止位数（The number of stop bits to use）
# bytesize=8, # 序列化消息字节大小（The bytesize of the serial messages）
# parity="N", # 使用哪种奇偶校验
baudrate=Args.baudrate, # 用于串行设备的波特率
# handle_local_echo=False, # 处理USB-to-RS485适配器的本地echo(Handle local echo of the USB-to-RS485 adaptor)
# ignore_missing_slaves=True, # ignore request to a missing slave
# broadcast_enable=False, #
# strict=True, # 使用严格的计时，针对Modbus RTU 为t1.5（use strict timing, t1.5 for Modbus RTU）
)
elif Args.comm == 'tls':
address = (Args.host if Args.host else '', Args.port if Args.port else None)
server = await StartAsyncTlsServer(
context=Args.context, # Data storage
host='localhost', # 定义用于连接的tcp地址
# port=port, # tcp监听端口
identity=Args.identity, # server identify
# custom_functions=[], # allow custom handling
address=address,
framer=Args.framer,
certfile=Args.cert_file_path, # The cert file path for TLS (used if sslctx is None)
# sslctx=sslctx, # The SSLContext to use for TLS (default None and auto create)
keyfile=Args.key_file_path, # The key file path for TLS (used if sslctx is None)
# password="none", # 用于解密私钥文件的密码
# ignore_missing_slaves=True,
# broadcast_enable=False,
# timeout=1
)
return server
async def async_helper():
_logger.info("Starting server...")
await run_async_server()
if __name__ == "__main__":
asyncio.run(async_helper(), debug=True)

复制代码

异步客户端代码实现

client_async.py

#!/usr/bin/env python3
'''Pymodbus异步客户端示例
'''
import asyncio
import logging
import pymodbus.client as modbusClient
from pymodbus import ModbusException
_logger = logging.getLogger('logger')
class Args:
comm = 'tcp' # 通讯模式，可选值 tcp、udp serial、tls
comm_defaults = {
'tcp': ['socket', 5020],
# 'tcp': ['rtu', 5020], # 如果采用ModbusRTU协议则使用这个
'udp': ['socket', 5020],
# 'serial': ['rtu', '/dev/ptyp0'], # Linux
'serial': ['rtu', 'COM2'] # Windows(本例中COM1和COM2是用Virtual Serial Port Driver 6.9软件成对添加的虚拟端口)
'tls': ['tls', 5020]
}
framer = comm_defaults[comm][0] # 帧类型
port = comm_defaults[comm][1] # 站点端口
host = '127.0.0.1' # 服务端地址
baudrate = 9600 # 串行设备波特率
timeout = 10 # 客户端访问服务器超时时间(该参数仅用于客户端(slave节点)),float型
# sslctx=sslctx, # 用于TLS的SSLContext (默认为None, 自动创建)
cert_file_path = './certificates/pymodbus.crt' # 用于TLS 证书文件路径 (如果sslctx为None时使用该文件)
key_file_path = './certificates/pymodbus.key' # 用于TLS 私钥文件路径 (如果sslctx为None时使用该文件)
async def run_async_client(modbus_calls=None):
"""Run sync client."""
_logger.info("### Create client object")
if Args.comm == "tcp":
client = modbusClient.AsyncModbusTcpClient(
Args.host,
port=Args.port, # on which port
# Common optional parameters:
framer=Args.framer,
timeout=Args.timeout,
retries=3,
reconnect_delay=1,
reconnect_delay_max=10,
# retry_on_empty=False,
# TCP setup parameters
# source_address=("localhost", 0),
)
elif Args.comm == "udp":
client = modbusClient.AsyncModbusUdpClient(
Args.host,
port=Args.port,
# Common optional parameters:
framer=Args.framer,
timeout=Args.timeout,
# retries=3,
# retry_on_empty=False,
# UDP setup parameters
# source_address=None,
)
elif Args.comm == "serial":
client = modbusClient.AsyncModbusSerialClient(
Args.port,
# Common optional parameters:
# framer=ModbusRtuFramer,
timeout=Args.timeout,
# retries=3,
# retry_on_empty=False,
# Serial setup parameters
baudrate=Args.baudrate,
# bytesize=8,
# parity="N",
# stopbits=1,
# handle_local_echo=False,
# strict=True,
)
elif Args.comm == "tls":
client = modbusClient.AsyncModbusTlsClient(
Args.host,
port=Args.port,
# Common optional parameters:
framer=Args.framer,
timeout=Args.timeout,
# retries=3,
# retry_on_empty=False,
# TLS setup parameters
sslctx=modbusClient.AsyncModbusTlsClient.generate_ssl(
certfile=Args.cert_file_path,
keyfile=Args.key_file_path,
# password="none",
),
server_hostname="localhost",
)
_logger.info("### Client starting")
await client.connect()
assert client.connected
if modbus_calls:
await modbus_calls(client)
client.close()
_logger.info("### End of Program")
async def run_a_few_calls(client):
try:
# 读取线圈状态
rr = await client.read_coils(0, 1, slave=0) # 从 0x00 地址开始读取1个线圈
print(rr.bits) # 输出：[True, False, False, False, False, False, False, False]
# assert len(rr.bits) == 8
print(rr) # 输出：ReadCoilsResponse(8)
# 读输入离散量
rr = await client.read_holding_registers(0, 1, slave=0) # 从 0x00 地址开始读取1个线圈
print(rr.registers) # 输出：[17]
rr = await client.read_holding_registers(4, 2, slave=0) # 从0x04 地址开始读取2个线圈
# assert rr.registers[0] == 17
# assert rr.registers[1] == 17
print(rr.registers) # 输出：[17, 17]
# 读保持寄存器
rr = await client.read_holding_registers(5, 4) # 从 0x05 地址开始读取4个线圈
print(rr.registers) # 输出：[17, 17, 17, 17]
# 读输入寄存器
rr = await client.read_input_registers(0x0F, 3, slave=0) # 从 0x0F 地址开始读取4个线圈
print(rr.registers) # 输出：[17, 17, 17]
rr = await client.read_input_registers(15, 3, slave=0) # 从 0x0F 地址开始读取4个线圈
print(rr) # 输出：ReadInputRegistersResponse (3)
print(rr.registers) # 输出：[17, 17, 17]
# 写单个线圈
rr = await client.write_coil(9, False, slave=0) # 将布尔值False写入 0x09 地址
print(rr) # 输出：WriteCoilResponse(9) => False
rr = await client.read_coils(9, 1, slave=0)
print(rr.bits) # 输出：[False, False, False, False, False, False, False, False]
# 写多个线圈
await client.write_coils(10, False, slave=0) # 将布尔值False写入 0x0A 地址
rr = await client.read_coils(10, 1, slave=0)
print(rr.bits) # 输出：[False, False, False, False, False, False, False, False]
rr = await client.read_coils(11, 1, slave=0)
print(rr.bits) # 输出：[True, False, False, False, False, False, False, False]
await client.write_coils(10, [False, False], slave=0) # 将布尔值False写入 0x0A 0x0B 地址
rr = await client.read_coils(11, 1, slave=0)
print(rr.bits) # 输出：[False, False, False, False, False, False, False, False]
# 写单个保持寄存器
await client.write_register(12, 0x0F, slave=0) # 将0x0F写入 0x0C 地址
rr = await client.read_input_registers(12, 3, slave=0)
print(rr.registers) # 输出：[17, 17, 17]
rr = await client.read_holding_registers(12, 4)
print(rr.registers) # 输出：[15, 17, 17, 17]
# 写多个保持寄存器
await client.write_registers(13, 0x0F, slave=0) # 将0x0F写入 0x0D 地址
rr = await client.read_holding_registers(13, 2)
print(rr.registers) # 输出：[15, 17]
await client.write_registers(13, [0x0F, 0x0E], slave=0) # 将0x0F写入 0x0D,0x0E 地址
rr = await client.read_holding_registers(13, 2)
print(rr.registers) # 输出：[15, 14]
except ModbusException:
pass
async def main():
await run_async_client(modbus_calls=run_a_few_calls)
if __name__ == "__main__":
asyncio.run(main(), debug=True)

复制代码