Omniplc

介绍

omniplc:一个面向多品牌、多协议 PLC 的 Python 统一通信库。一次编写,即可通过一致的 API 对接三菱、欧姆龙、基恩士、汇川、松下、丰田、罗克韦尔(AB)、倍福(TwinCAT)、西门子(S7)等 PLC/扫码枪、OPC-UA 服务器与 CNC 机床(MTConnect),支持 Modbus、MC(3E/4E/1E 以太网帧、1C/3C/4C 串口帧)、FINS、NJ/NX CIP(EtherNet/IP)、KV Host Link、KV MC 协议兼容(SLMP)、汇川 H3U/H5U(Modbus TCP/RTU、MC 协议兼容 3E)、松下 FP(MC 协议兼容 3E、MEWTOCOL)、SR、TOYOPUC 计算机链接、EtherNet/IP(Logix 标签读写)、TwinCAT ADS(封装 pyads)、西门子 S7(封装 python-snap7,DB/I/Q/M)、通用自定义 TCP(分隔符成帧)、OPC-UA、MTConnect 数采等协议。

• Python 3.7.9+,uv 开发,核心零第三方依赖
• 命名与使用习惯对齐,迁移成本极低
• 全量类型标注(PEP 484 + py.typed),mypy 检查通过
• 内置全局报文调试开关(omniplc.set_debug(True) 一键输出所有协议的请求/响应报文)
• 线程安全、惰性自动重连、可配置超时/重试
• 同步 + 异步(异步类 = 同步类名前加 A)双轨 API

类继承图

BaseClient(ABC,模板方法:连接状态机 / 事务锁 / 惰性重连 / 类型化读写只写一次)
│
├── ModbusBaseClient —— 寄存器级公共逻辑:字序 / 类型分发 / 范围校验
│   ├── ModbusTcpClient —— MBAP over TCP(502)
│   │   └── InovanceTcpClient —— 汇川 H3U/H5U,继承 Modbus 只换软元件地址映射
│   └── ModbusRtuClient —— 站号+PDU+CRC16 over 串口(需 pyserial)
│       └── InovanceRtuClient —— 汇川 H3U/H5U RTU(9600-8N2),继承 Modbus 只换地址映射
│
├── MelsecMcTcpClient —— 三菱 MC 3E/4E/1E 帧 over TCP(2000)
│   ├── KeyenceMcTcpClient —— 基恩士 KV MC 兼容 / SLMP 3E(5000),只换码表
│   ├── InovanceMcTcpClient —— 汇川 MC 兼容 3E,换码表 + 记号换算(S→L、R=D+8000、X/Y 八进制)
│   └── PanasonicMcTcpClient —— 松下 FP0H/FP7 MC 兼容 3E,换码表 + 记号换算(字号×16+位号、R9000+→SM、D90000+→SD)
├── MelsecMcUdpClient —— 三菱 MC 同帧型 over UDP(2000)
│   └── KeyenceMcUdpClient —— 基恩士 KV MC 兼容 SLMP 3E over UDP(5000),与 TCP 版共用码表覆写
├── MelsecMcSerialClient —— 三菱 MC 串口帧(C24):1C A 兼容 ASCII 格式 4 / 3C ASCII 格式 4 / 4C 二进制格式 5,需 pyserial
├── MelsecMxClient —— 三菱 MX Component(Windows,comtypes,逻辑站号)
│
├── OmronFinsTcpClient —— 欧姆龙 FINS + TCP 握手(9600)
├── OmronFinsUdpClient —— 欧姆龙 FINS over UDP(9600)
├── AllenBradleyEthIpClient —— 罗克韦尔 AB EtherNet/IP(44818):Logix 标签自描述,unconnected/connected 双通道
│   └── OmronCipClient —— 欧姆龙 NJ/NX CIP(44818):unconnected 直发无背板路由,NJ 变量读写
├── BeckhoffAdsClient —— 倍福 TwinCAT ADS(AMS 851,封装 pyads):变量名即地址,ADSError 不断线
├── KeyenceHostLinkTcpClient —— 基恩士 KV Host Link over TCP(8000)
├── KeyenceHostLinkUdpClient —— 基恩士 KV Host Link over UDP(8000)
├── KeyenceSrClient —— 基恩士 SR 扫码枪 TCP(9004,LON/LOFF 触发扫码)
├── PanasonicMewtocolTcpClient —— 松下 MEWTOCOL over TCP(1024):ASCII 帧 + BCC,RCS/WCS 单接点、RD/WD 数据区
├── PanasonicMewtocolUdpClient —— 松下 MEWTOCOL over UDP(1024)
├── ToyopucTcpClient —— 丰田 TOYOPUC 计算机链接 over TCP(1025)
├── ToyopucUdpClient —— TOYOPUC 同帧 over UDP(1025)
├── OpcUaClient —— OPC-UA opc.tcp 会话(4840,封装 asyncua)
├── MTConnectClient —— CNC 机床数采(HTTP/XML 只读,Agent 默认 5000)
├── SiemensS7Client —— 西门子 S7(102,rack/slot 路由,封装 python-snap7)
└── OpenTcpClient —— 通用自定义 TCP/IP(端口按设备):分隔符/定长成帧 + 内部缓冲,send/receive/transact*

异步镜像(omniplc.aio):类名 = 同步类名前加 A,签名同名同型,共 28 个
AModbusTcpClient / AModbusRtuClient / AInovanceTcpClient / AInovanceRtuClient / AInovanceMcTcpClient
AMelsecMcTcpClient / AMelsecMcUdpClient / AMelsecMcSerialClient / AMelsecMxClient / AOmronFinsTcpClient
AOmronFinsUdpClient / AOmronCipClient / ABeckhoffAdsClient / AAllenBradleyEthIpClient / AKeyenceHostLinkTcpClient
AKeyenceHostLinkUdpClient / AKeyenceMcTcpClient / AKeyenceMcUdpClient / APanasonicMcTcpClient / APanasonicMewtocolTcpClient
APanasonicMewtocolUdpClient / AKeyenceSrClient / AToyopucTcpClient / AToyopucUdpClient / AOpcUaClient
AOpenTcpClient / AMTConnectClient / ASiemensS7Client

原生异步(omniplc.native):独立层,类名 = 同步类名前加 Async,首批 5 个
AsyncBaseClient(异步基类,事务模板与同步层同口径)
AsyncModbusTcpClient
AsyncMelsecMcTcpClient / AsyncMelsecMcUdpClient(1E/3E)
AsyncOmronFinsTcpClient / AsyncOmronFinsUdpClient

详细架构设计见 docs/architecture.md。

安装

uv add omniplc            # 或 pip install omniplc
uv add 'omniplc[serial]'  # 需要 Modbus RTU 或三菱 MC 串口 1C/3C/4C 帧时(pyserial)
uv add 'omniplc[mx]'      # 需要三菱 MX Component(Windows)时
uv add 'omniplc[opcua]'   # 需要 OPC-UA 时(安装 asyncua)
uv add 'omniplc[ads]'     # 需要倍福 TwinCAT ADS 时(安装 pyads,Windows 需 TcAdsDll 运行库)
uv add 'omniplc[s7]'      # 需要西门子 S7 时(安装 python-snap7,按解释器版本自动二选一)

快速上手

from omniplc import ModbusTcpClient, DataType

client = ModbusTcpClient(ip_address="192.168.0.10", port=502, station=1)
client.receive_timeout = 3.0        # 超时走属性,不进构造函数
client.connect()

ok, value = client.read_float("hr100")     # 读返回 (是否成功, 值)
ok = client.write_float("hr100", 3.14)     # 写返回 bool
print(client.last_error)                   # 失败原因在这里

client.disconnect()

# 上下文管理器:进入自动连接,失败抛 ConnectionError
with ModbusTcpClient("192.168.0.10", 502, 1) as client:
    results = client.read_many(["hr0", "hr2"], DataType.FLOAT)  # [(是否成功, 值), ...] 逐点列表
    ok, value = results[0]

# 通用 read/write 推荐传 DataType 枚举(IDE 自动补全),也兼容字符串
ok, value = client.read("hr0", DataType.FLOAT)

# 掩码写(FC22):设备侧原子位修改,替代读-改-写两段事务
ok = client.write_mask_register("hr100", and_mask=0xFFFE, or_mask=0x0001)

# 读写多寄存器(FC23):单事务「先写后读」,读到的是写入生效后的值
ok, values = client.read_write_registers("hr200", 2, "hr100", [0x0001, 0x0002])

# 设备标识(FC43/14):厂商名/产品代码/版本号等;More Follows 自动翻页
ok, info = client.read_device_id()
print(info["vendor_name"], info["product_code"], info["major_minor_revision"])
ok, raw = client.read_device_object(0x02)   # 单个对象(个体访问),返回原始字节

# Modbus RTU(串口)
from omniplc import ModbusRtuClient
rtu = ModbusRtuClient(station=1)
rtu.configure_serial("COM3", baud_rate=9600)

三菱 / 欧姆龙 / 基恩士 / 汇川 / 丰田

from omniplc import MelsecMcTcpClient, OmronFinsUdpClient, McFrame

# 三菱 MC:frame=McFrame.FRAME_3E/FRAME_4E(QnA 兼容)或 FRAME_1E(A 兼容,A 系列)
mc = MelsecMcTcpClient(ip_address="192.168.3.39", port=2000, frame=McFrame.FRAME_3E)
mc.connect()
ok, value = mc.read_ushort("D100")
ok = mc.write_bool("M100", True)
ok, values = mc.read_batch([("D100", "short"), ("M100", "bool")])  # 0406 多块批量读,单事务

# 三菱 MC 串口帧(C24 串口模块,需 pyserial):1C=A 兼容 ASCII 格式 4(BR/WR/BW/WW),
# 3C=QnA 兼容 ASCII 格式 4,4C=二进制格式 5
# 软元件地址与 3E 帧一致;串口参数须与 C24"传送设定"一致,默认访问连接站 CPU(PC 号 FF)
from omniplc import MelsecMcSerialClient
mc_sio = MelsecMcSerialClient(frame=McFrame.FRAME_4C)
mc_sio.configure_serial("COM3", baud_rate=9600)
mc_sio.connect()
ok, value = mc_sio.read_ushort("D100")
ok = mc_sio.write_bool("M100", True)

# 三菱 MX Component(Windows):通信参数在通信设置实用程序中配置为逻辑站号
# 安装:pip install 'omniplc[mx]'
from omniplc import MelsecMxClient
mx = MelsecMxClient(logical_station_number=1)
mx.connect()
ok, value = mx.read_ushort("D100")
ok = mx.write_bool("M10", True)
# ReadDeviceRandom 原生随机读:仅 16 位类型(BOOL/SHORT/USHORT),单事务
ok, values = mx.read_batch([("M10", "bool"), ("D100", "short")])

# 欧姆龙 FINS:TCP 自动做节点分配握手,UDP 无握手
fins = OmronFinsUdpClient(ip_address="192.168.250.1", port=9600)
fins.connect()
ok, value = fins.read_ushort("D100")
ok = fins.write_bool("CIO0.5", True)
ok, values = fins.read_batch([("D100", "short"), ("CIO0.5", "bool")])  # 0104 多存储区读,单事务

# 欧姆龙 NJ/NX CIP(内置 EtherNet/IP):Sysmac 变量自描述,地址即变量名
# TestVar / MyArray[5] / Motor[2].Speed;继承 AB 客户端,直发不包 UC Send
from omniplc import OmronCipClient
nj = OmronCipClient(ip_address="192.168.0.10", port=44818)
ok, value = nj.read_int("TestVar")
ok = nj.write_bool("RunFlag", True)
nj_c = OmronCipClient(ip_address="192.168.0.10", connected_messaging=True)  # 连接型

# 倍福 TwinCAT(ADS):变量名即地址(MAIN.nCounter / .gGlobal / GVL.MyVar)
# 安装:pip install 'omniplc[ads]'
# Windows 侧还需 Beckhoff TcAdsDll 运行库,随 TwinCAT ADS 安装
from omniplc import BeckhoffAdsClient
bc = BeckhoffAdsClient(ip_address="192.168.0.10", ads_port=851)  # net_id 默认 IP+.1.1
ok, value = bc.read_int("MAIN.nCounter")
ok = bc.write_bool("MAIN.bStart", True)
ok, text = bc.read_string("MAIN.sRecipe")

# 罗克韦尔 AB EtherNet/IP:Logix 标签自描述,类型不符会明确报错
# 地址即标签名:MyDint / MyArray[5] / MyUdt.Member / MyDint.3(位)/ 程序作用域 Program:prog.Tag
from omniplc import AllenBradleyEthIpClient
ab = AllenBradleyEthIpClient(ip_address="192.168.1.20", port=44818, slot=0)
ab.connect()
ok, value = ab.read_int("MyDint")
ok = ab.write_bool("StartCmd", True)
ok, text = ab.read_string("RecipeName")
# 0x0A 多服务包:一帧混读多个标签(上限 32 条;BOOL 首次批量读会做一次类型发现)
ok, values = ab.read_batch([("MyDint", "int"), ("MyReal", "float"), ("RecipeName", "string")])
ok, info = ab.get_plc_info()      # Identity Object:厂商/序列号/产品名
ok, ident = ab.list_identity()    # ENIP 单播发现
# connected 消息(Forward Open + SendUnitData,大批量轮询吞吐更高):
ab_c = AllenBradleyEthIpClient(ip_address="192.168.1.20", connected_messaging=True)

# 基恩士 KV Host Link:ASCII 行式协议,地址如 DM100 / R515 / W100
from omniplc import KeyenceHostLinkTcpClient
kv = KeyenceHostLinkTcpClient(ip_address="192.168.0.10", port=8000)
kv.connect()
ok, value = kv.read_ushort("DM100")
ok = kv.write_bool("R515", True)

# 基恩士 KV MC 协议兼容(SLMP):二进制 3E 帧复用三菱实现,仅码表不同
# 地址如 R5(位,十进制)/ DM100(字,十进制)/ B1F / W10(十六进制)/ ZR100
from omniplc import KeyenceMcTcpClient, KeyenceMcUdpClient
kvmc = KeyenceMcTcpClient(ip_address="192.168.1.22", port=5000)
ok, value = kvmc.read_ushort("DM100")
ok = kvmc.write_bool("R5", True)
kvmc_u = KeyenceMcUdpClient(ip_address="192.168.1.22", port=5000)  # UDP 走线,一问一答一数据报

# 汇川 H3U/H5U:Modbus TCP/RTU + 汇川软元件地址映射
# 地址如 D100 / R100 / M10 / SM10 / X17(八进制)/ D100.3;T/C 位=接点、字=当前值
from omniplc import InovanceTcpClient, InovanceRtuClient
h3u = InovanceTcpClient(ip_address="192.168.1.88", port=502, station=1)
ok, value = h3u.read_ushort("D100")
ok = h3u.write_bool("M10", True)
rtu2 = InovanceRtuClient(station=1)
rtu2.configure_serial("COM3")   # 汇川缺省 9600-8N2

# 汇川 MC 协议兼容(3E 帧,Easy 系列/H5U 固件 V6.4.0.0+ 的"MC配置"功能)
# 帧按三菱口径编码;S 按三菱 L 码、R 与 D 统一编址(R100=D8100)、X/Y 八进制命名
# 端口无出厂默认,须与 AutoShop"MC配置"中设置一致
from omniplc import InovanceMcTcpClient
imc = InovanceMcTcpClient(ip_address="192.168.1.88", port=2000)
ok, value = imc.read_ushort("R100")
ok = imc.write_bool("X17", True)

# 松下 FP0H/FP7:MC 协议兼容(3E 帧,仅二进制成批读/写)+ MEWTOCOL(TCP/UDP)
# MC:位软元件按"字号+位号"(R1F=字1位F/R1.15),R9000+(字900)起为 SM,D90000+ 为 SD
# 端口以模块配置为准;MEWTOCOL 接点 X/Y/R/T/C/L + 数据 D/L/F/S/K(K=经过值,S=设定值)
from omniplc import PanasonicMcTcpClient, PanasonicMewtocolTcpClient, PanasonicMewtocolUdpClient
pmc = PanasonicMcTcpClient(ip_address="192.168.0.10", port=2000)
ok, value = pmc.read_ushort("D100")
ok = pmc.write_bool("R1F", True)
mew = PanasonicMewtocolTcpClient(ip_address="192.168.0.10", port=1024, station=1)
ok, value = mew.read_ushort("D100")
ok = mew.write_bool("Y0.3", True)
mewu = PanasonicMewtocolUdpClient(ip_address="192.168.0.10", port=1024)

# 基恩士 SR 扫码枪:触发式设备,scan() 返回 (是否读到, 条码文本)
from omniplc import KeyenceSrClient
sr = KeyenceSrClient(ip_address="192.168.0.10", port=9004, scan_dwell=1.0)
sr.connect()
ok, code = sr.scan()        # LON → 窗口 → LOFF → 读应答

# 丰田 TOYOPUC 计算机链接:二进制帧,地址如 D0100 / M0201 / X0010H / M0201W
from omniplc import ToyopucTcpClient
toyopuc = ToyopucTcpClient(ip_address="192.168.0.10", port=1025)
ok, value = toyopuc.read_ushort("D0100")   # 编号为十六进制
ok = toyopuc.write_bool("M0201", True)     # 位软元件 CMD=20/21 直读直写

# OPC-UA:标准 NodeId 寻址,读写按显式数据类型编解码;入口同其他客户端(IP+端口)
# 安装:pip install 'omniplc[opcua]'
from omniplc import OpcUaClient
opc = OpcUaClient("192.168.0.10", 4840)
ok, value = opc.read_float("ns=2;s=Device.Temperature")
ok = opc.write_ushort("ns=2;s=Device.Speed", 1200)

# 通用自定义 TCP:任意分隔符成帧设备(称重仪表、传感器、自定义程序等)
# 分隔符/编码/收发行为构造期可配;超时与重连沿用属性(client.receive_timeout / client.retries)
from omniplc import OpenTcpClient
dev = OpenTcpClient(ip_address="192.168.0.10", port=9000, delimiter="\r\n")
dev.receive_timeout = 2.0
dev.connect()
ok = dev.send_text("READ")             # 自动补分隔符(append_delimiter 可关)
ok, raw = dev.receive()                # 按分隔符收一帧(bytes),跨分片自动拼接
ok, text = dev.transact_text("VER")    # 发送并收一帧(str);坏帧/解码失败断线重连,超时不断线

# CNC 机床数采(MTConnect):数据项 id 即地址,Agent 默认端口 5000(标准库实现,零第三方依赖)
# FANUC/三菱等控制器经适配器喂给 Agent 即可采;先 snapshot() 查看机器实际提供的数据项
from omniplc import MTConnectClient
cnc = MTConnectClient("192.168.0.10", 5000)
cnc.connect()
ok, speed = cnc.read_float("Sspeed")   # 主轴转速(文本值自动转 float)
ok, program = cnc.read_string("program")
ok, items = cnc.snapshot()             # 全量当前值快照 {数据项 id: 文本值}
ok, alarms = cnc.read_conditions()     # 条件项(Fault/Warning/Normal)列表
ok, device = cnc.probe()               # 设备信息(name/uuid 等)

# 西门子 S7(封装 python-snap7):DB/I/Q/M 绝对寻址,ISO-on-TCP 102,rack/slot 路由
# S7-1200/1500 需勾选"允许来自远程对象的 PUT/GET 通信访问",DB 须为非优化块
# 安装:pip install 'omniplc[s7]'
# 依赖按 Python 版本自动二选一:
#   3.7~3.9 → python-snap7 1.3(C 封装,64 位用捆绑库;32 位需自备 snap7.dll 经 dll_path 指定)
#   3.10+   → python-snap7 3.x(纯 Python 实现,无需原生 DLL)
from omniplc import SiemensS7Client
s7 = SiemensS7Client("192.168.0.1", rack=0, slot=1)  # 300/400 的 CPU 常在槽位 2
s7.connect()
ok, temp = s7.read_float("DB1.DBD6")   # DB 双字起点,REAL
ok = s7.write_bool("DB1.DBX0.3", True) # DB 位(锁内读-改-写)
ok, current = s7.read_ushort("MW10")   # Merker 字
ok, text = s7.read_string("DB1.DBS20", length=32)  # S7 String(头 2 字节声明/实际长)

批量读取(默认逐点 / 协议原生单事务)

read_many(地址列表, 数据类型) 的默认契约是逐点独立容错:基类逐点发出独立事务,单点失败不影响其他点。read_batch([(地址, 类型), ...]) 是协议级批量入口,仅在支持单事务批量的驱动上提供。以下驱动把批量入口覆写为协议级单事务——一帧往返读回多个点(不是循环单点),是整批语义:任一地址非法或设备拒绝则整批失败(原因在 last_error;要逐点容错请逐点 read):

• 三菱 MC 3E/4E:0406 多块批量读(混软元件,总块数 ≤120);KV/汇川/松下 MC 兼容子类经继承同享
• 欧姆龙 FINS:0104 多存储区读(以太网上限 167 条)
• AB / 欧姆龙 NJ-NX CIP:0x0A 多服务包(上限 32 条;BOOL 首次批量读做一次类型发现后缓存)
• OPC-UA:UA Read 服务原生多节点(asyncua read_values 单请求);任一节点非法或服务端拒绝则整批失败,原因进 last_error,需要逐点容错请逐点 read
• MX Component:ReadDeviceRandom(软元件列表换行分隔;仅 16 位类型)
• Modbus:read_batch/write_batch 按 (区域, 类型) 分组、组内连续地址合并为单条 FC(读 01/02/03/04,写 15/16)——Modbus 协议不支持跨 FC 单事务,故为 K 笔而非 1 笔(K ≤ 地址数,典型 1 笔);read_write_registers(FC23)可在一个事务内先写后读

# 混类型混软元件:一帧取回全部值(以 MC 为例,其余驱动同款接口)
ok, values = mc.read_batch([
    ("D100", "short"), ("D102", "float"), ("M100", "bool"), ("D110.3", "bool"),
])
# 统一类型批量:read_many(等价于 read_batch 的同类型版本)
ok_list = mc.read_many(["D0", "D2", "D4"], DataType.FLOAT)

报文调试(全局开关)

import omniplc

omniplc.set_debug(True)   # 之后所有协议客户端输出请求/响应报文(十六进制)
omniplc.set_debug(False)  # 关闭

• 走线型协议(TCP/UDP/串口)在传输层统一挂钩,输出每次收发的原始字节; TCP 应答分多段到达时按段输出。输出形如 tcp://192.168.0.10:2000 → 发送 12B: 50 00 00 FF …(单条最多转储 4096B)
• 会话型协议(OPC-UA / ADS / MX Component 无字节流)输出操作级日志, 如 ads://192.168.0.10.1.1:851 读 MAIN.rTemp(PLCTYPE_REAL) → 3.14
• 输出走 logging(记录器名 omniplc.debug,DEBUG 级):应用已配置 logging 时自动汇入既有日志体系;未配置时自动挂 stderr 处理器,开箱即用
• 进程级开关,同步与异步客户端共用;连接建立/断开也会输出,便于观察惰性重连

异步(两套:包装层 omniplc.aio / 原生层 omniplc.native)

异步客户端是多设备并发的手段,不是单连接提速(单设备逐笔轮询用同步即可,异步只多线程切换开销): 协议事务在单连接内本就是"一问一答"串行,收益来自把多台设备的等待重叠——10 台设备并发采集约等于顺序轮询的 1/10 耗时。

选哪套(先读再选型):

两套都保留:包装层继续覆盖全部协议(过渡期用),原生层逐批补齐;同一个 import asyncio 项目里可以混用(互不影响)。

包装层 omniplc.aio 的实现方式与边界:它不是原生 asyncio 协议栈, 每个 await 把同步调用投递到该客户端自己的单工作线程,协议编解码只有一份代码。 由此有三条硬边界:

• 同一客户端仍是串行的:协议调用在工作线程里按 FIFO 排队(与同步侧同一把 事务锁),并发不会让单台设备变快;收益只来自跨设备重叠等待,需要单设备并发 吞吐请多开实例(连接池留 v1.x)。
• 事件循环不被 I/O 阻塞,但同步属性是直读:await 期间 I/O 在工作线程; 而 connected / last_error* / stats / receive_timeout 读写是同步直读 同步实例——不发报文、不切线程,读到的可能不是原子快照。
• 没有 asyncio 原生取消:asyncio.wait_for 超时只是放弃等待,已提交的 写事务仍会在工作线程里跑完(写动作不可回滚);要硬性限时请用同步侧的 receive_timeout / 事务 deadline。close() 则相反:它会排空已提交任务 (不锯断在途事务)再释放线程,因此关闭后不会再有帧落线。

import asyncio
from omniplc.aio import AMelsecMcTcpClient, AOmronFinsTcpClient, AAllenBradleyEthIpClient

async def main():
    clients = [
        AMelsecMcTcpClient("192.168.0.11", 2000),
        AOmronFinsTcpClient("192.168.0.12", 9600),
        AAllenBradleyEthIpClient("192.168.0.13", 44818),
    ]
    for client in clients:
        await client.connect()
    # 三台设备的同时刻采集:总耗时 ≈ 最慢一台的往返,而非三者之和
    d1, d2, d3 = await asyncio.gather(
        clients[0].read_float("D100"),
        clients[1].read_float("D100"),
        clients[2].read_float("MyReal"),
    )
    for client in clients:
        await client.disconnect()

asyncio.run(main())

原生层 omniplc.native(方法名与同步版同名同型,await 即可):

import asyncio
from omniplc.native import AsyncModbusTcpClient

async def main():
    # 支持 async with:失败抛 ConnectionError,退出自动断开
    async with AsyncModbusTcpClient("192.168.0.10", 502, 1) as client:
        ok, value = await client.read_float("hr100")
        print(client.stats["transactions"])     # 属性是同步读取,不阻塞事件循环

asyncio.run(main())

• 同一客户端串行、多客户端真并发(与包装层一致):同一实例的协议调用 排在一把 asyncio.Lock 后面,跨实例天然并行。
• 取消是真中断:asyncio.wait_for 超时会取消事务;若请求已发出,连接按 "链路可能残留未配对应答"保守拆连重同步(下次事务惰性重连),仅排队未发出 则保持连接。
• 超时口径与同步层一致:TCP 读超时按连接死亡拆连;UDP 接收超时不拆连 (数据报整收,无残渣)、UDP 发送超时(本地缓冲打满)按 OSError 语义拆连 ——与同步 UdpTransport 的收/发两条路径逐条对应。
• 一个实例绑定一个事件循环(事务锁按首次使用时的循环惰性创建,模块级 构造再 asyncio.run 也正常);跨循环/跨线程共享同一实例不支持。
• 能力面:单点读写 + 类型化方法 + read_string/write_string + 点位表; 批量(read_many/read_batch)与各家扩展方法(FC 22/23/43、MC 0406、 FINS 0104 等)尚未进入原生层,需要时用包装层或同步客户端。

点位表(可选)

from omniplc import TagTable

# tags.json 格式:[{"tag_id": "furnace_temp", "address": "D100", "data_type": "float",
#                  "scale": 0.1, "remark": "炉温"}, ...]
# tag_id 为程序用标识(一般字母/数字),remark 为中文备注(供人员查看记录,可省略)
client.bind_tags(TagTable.from_json("tags.json"))
ok, value = client.read_tag("furnace_temp")   # 点位标识 → 地址+类型,自动应用缩放

迁移(v0.33.0 schema 破坏性变更):旧表 name 字段的值原样迁入 tag_id (点位标识,项目内唯一,一般字母/数字),原中文说明改写到新增的 remark (可选,缺省为空),其余字段不变;字段语义见 tag.py 的 Tag 文档。

错误处理约定

读返回 (bool, 值),写返回 bool,不抛自定义异常; 失败原因记录在 client.last_error(含 PLC 原始错误码)。参数非法(地址/类型/ 范围错误)抛 ValueError。read_many/write_many 默认逐点独立容错,单点失败不影响其他点; 被覆写为协议级单事务的驱动(MC 0406 / FINS 0104 / AB 0x0A / OPC-UA UA Read / MX ReadDeviceRandom)为整批语义:任一点失败则整批失败,原因在 last_error, 要逐点容错请逐点 read。

失败分类与原始码(v0.34.0 起):除 last_error 文本外,另有两个只读 属性供上位系统分类告警——client.last_error_category(ErrorCategory 枚举:TRANSPORT/PROTOCOL/DEVICE/TIMEOUT/UNKNOWN)与 client.last_error_code(PLC 原始错误码,如 MC 结束码 0xC059、Modbus 异常码 2、FINS 结束码 0x2108;传输类错误为 errno,无码为 None)。 成功读写后三者一并清空:

from omniplc import ErrorCategory

ok, value = client.read_ushort("D100")
if not ok:
    cat = client.last_error_category
    if cat is ErrorCategory.DEVICE:
        ...  # PLC 报错:看 last_error_code 区分地址越界/功能不支持
    elif cat is ErrorCategory.TRANSPORT:
        ...  # PLC 不可达:最高级告警
    elif cat is ErrorCategory.TIMEOUT:
        ...  # 链路完好但超时:查负载/串扰/超时配置

OPC-UA 推模式(v0.35.0 起)

OPC-UA 除拉模式读写外,还支持服务端推数据与事件:

• subscribe_data_change(node, on_change, sampling_interval_ms=1000) 订阅数据 变化(回调 (value, node_id, source_timestamp),回调异常只记日志与 last_error,不杀订阅)、subscribe_event(node, on_event, event_filter=None) 订阅事件(EventFilter 可选透传)
• OpcUaSubscription 订阅句柄(unsubscribe() 幂等),active_subscriptions 给出活跃订阅快照;disconnect() 先退订再断开
• browse(node_text="Root", recursive=True, max_depth=None) 枚举地址树, 返回 {node_id: {browse_name, node_class, children}} 嵌套字典
• 断线不自动重订,订阅重建策略留调用端;同步与 aio 异步双轨可用

连接退避(v0.34.0 起)

连接失败(建连或握手)后自动进入指数退避门控:第 n 次连续失败后, 下一次 connect() 在 uniform(0, min(0.5 × 2ⁿ, 30)) 秒内被直接拒绝 (返回 False,last_error 提示"连接退避中")——时间戳比较,不发包、 不 sleep,PLC 断电/网线松动时紧密轮询的调用方不再形成高频重连风暴。 连接成功或显式 disconnect() 后门控与失败计数全部重置。

• 默认开启;client.reconnect_backoff = False 一行恢复 v0.33 行为
• client.next_connect_in:距下次允许连接的剩余秒数(None = 可立即连)
• 门控拒绝不计入 stats["error_count"](无真实网络动作)
• 显式 disconnect() 会重置门控与失败计数(视为干净起点); 失败重试循环中请勿"先 disconnect 再 connect",否则退避不生效
• 配置了 retries 时,门控窗口内的重试直接结束(不空转), last_error 保留武装门控的那次真实失败根因

连接健康统计(v0.30.0 起)

所有 BaseClient 子类提供 client.stats 只读快照,返回类型 omniplc.ClientStats——TypedDict,字段名可被 IDE 与类型检查补全 (运行期就是普通 dict:3.7 无 typing.TypedDict,退化为 dict 子类, 取值方式与既有行为完全不变),且每次返回拷贝,改返回值不影响内部计数。 字段:

• connect_count / disconnect_count / transactions / error_count / device_error_count:计数器(锁内更新);device_error_count 只计 PLC 明确返回错误码的次数——接收超时(归 last_error_category = timeout) 与无码失败(能力缺失、设备侧条件)不计入,它们只进 error_count
• last_error_at / last_connect_at / last_success_at:time.monotonic() 时间戳(秒);成功/失败/建连时刷新;跨重启无意义,用于现场判断 "多久没成功/多久前出错"
• last_rtt:最近一次成功事务的往返耗时(秒,含 PLC 等待),微秒级开销

异步镜像 ABaseClient.stats 同步转发。判断示例:

s = client.stats
if s["error_count"] > 10 and (s["last_success_at"] or 0) < (time.monotonic() - 60):
    # 错误多且一分钟没成功过:报警/触发诊断
    ...

需要静态检查/补全时标注返回类型即可:

from omniplc import ClientStats

def dump(s: ClientStats) -> None:
    print(s["transactions"], s["last_rtt"])

dump(client.stats)   # 键名拼错、字段用错类型在 mypy/pyright 阶段即报

真机联测待做(v0.30.0 整理)

下表汇总散落各处的真机核证项(实现已完成,缺真机条件或排队中):

实际真机联测通过项的核验记录见 docs/real-machine-checklist.md(按厂商/协议/读写独立勾选)。

全协议 × 走线矩阵

表中 v1.x(...) 表示该走线留待 v1.x 版本实现,非版本号标注。

变更历史

按版本号降序的完整变更日志已迁出至 CHANGELOG.md(从 v0.40.1 到 v0.1 的详细说明);当前发布版本以 Git 标签为准(git tag -l 'v*')。

开发

uv sync --extra dev             # 安装开发依赖(dev extra 已含 comtypes,供静态检查解析)
uv run python -m pytest tests -q     # 测试(无需真机;32 位 py3.7 venv 的 exe shim 兼容性问题走 -m)
uvx ruff check src tests             # 代码检查
uvx mypy src/omniplc                 # 类型检查(python_version = 3.9,配置见 pyproject.toml)
uvx ty check src/omniplc             # ty 类型检查(Astral,配置见 pyproject.toml [tool.ty.src])

License

MIT