如何使用asyncua库精准控制tankFunction函数的扫描时间？

我测试Python的asyncua库，程序与本地PC上的OPC服务器通信，main_client()函数可读取值并修改另一值，功能正常。还添加了tankFunction函数模拟水箱液位在0到4之间升降。现在需要控制该函数的扫描时间，让tankFunction精准每秒运行一次，但当前脚本操作间隔约2秒，尝试过asyncio.sleep()和查阅asyncua文档仍未解决，对asyncio工作机制理解不足。

原代码：

import asyncio
from asyncua import Client, ua
import datetime

url = "opc.tcp://localhost:4840"
namespace = "CODESYSSPV3/3S/IecVarAccess"

#Parameters
global u 
global ykm1
u = 1
ykm1 = 0

def tankFunction(l,w,h):
    global u
    global ykm1

    y = 0.25*u+ykm1

    if y < 0:
        y = 0
        u=0.1
    elif y > 4:
        y = 4
        u = -0.1
    ykm1 = y
    return y

async def main_client():
    print(f"Connecting to {url} ...")
    async with Client(url=url) as client:
        node = client.get_node("ns=4;s=|var|CODESYS Control Win V3 x64.Application.GVL_Modbus.Mb_Tank01_ManualDensity")
        var = await node.get_value()
        print(var)
        
        node2 = client.get_node("ns=4;s=|var|CODESYS Control Win V3 x64.Application.GVL_Modbus.Tank_01_Active")
        await node2.write_value(True, ua.Boolean)

        now = datetime.datetime.now()
        print(now.time())

while True:
    asyncio.run(main_client())
    print(tankFunction(1,5,3))

终端输出：

Connecting to opc.tcp://localhost:4840 ...
0
09:37:54.791759
0.25
Connecting to opc.tcp://localhost:4840 ...
0
09:37:56.913719
0.5
Connecting to opc.tcp://localhost:4840 ...
0
09:37:59.039703
0.75

操作间隔约2.1秒，不符合需求。

问题分析 #

1. 重复创建连接与事件循环：每次while True循环都调用asyncio.run(main_client())，会重新初始化asyncio事件循环、重新连接OPC服务器，连接握手的耗时直接拉长了循环间隔。
2. 同步与异步逻辑混合：外层用同步while True循环嵌套异步任务，无法利用asyncio的异步调度精准控制定时。
3. 全局变量冗余：使用全局变量维护水箱状态，代码可读性和可维护性差。

解决方案 #

修改代码，复用OPC连接，将所有逻辑放到异步主循环中，用asyncio.sleep()结合时间补偿实现精准定时：

import asyncio
from asyncua import Client, ua
import datetime

url = "opc.tcp://localhost:4840"
namespace = "CODESYSSPV3/3S/IecVarAccess"

class TankSimulator:
    def __init__(self):
        self.u = 1
        self.ykm1 = 0

    def update_level(self, l, w, h):
        y = 0.25 * self.u + self.ykm1

        if y < 0:
            y = 0
            self.u = 0.1
        elif y > 4:
            y = 4
            self.u = -0.1
        self.ykm1 = y
        return y

async def main():
    tank = TankSimulator()
    # 只建立一次OPC连接
    async with Client(url=url) as client:
        print(f"Connected to {url}")
        # 预先获取节点，避免每次循环重复查找
        density_node = client.get_node("ns=4;s=|var|CODESYS Control Win V3 x64.Application.GVL_Modbus.Mb_Tank01_ManualDensity")
        active_node = client.get_node("ns=4;s=|var|CODESYS Control Win V3 x64.Application.GVL_Modbus.Tank_01_Active")
        # 先设置一次Active状态
        await active_node.write_value(True, ua.Boolean)

        while True:
            # 记录循环开始时间，用于时间补偿
            start_time = datetime.datetime.now()
            
            # 执行OPC读取操作
            var = await density_node.get_value()
            print(var)
            
            # 更新水箱液位并打印
            level = tank.update_level(1,5,3)
            print(level)
            
            # 打印当前时间
            print(start_time.time())
            
            # 计算需要sleep的时长：1秒减去本次循环已消耗的时间
            elapsed = (datetime.datetime.now() - start_time).total_seconds()
            sleep_time = max(0, 1 - elapsed)
            await asyncio.sleep(sleep_time)

if __name__ == "__main__":
    asyncio.run(main())

关键改进点 #

1. 复用OPC连接：将OPC连接放在外层异步循环中，只建立一次连接，避免重复连接的耗时。
2. 预先获取节点：提前查找OPC节点，避免每次循环重复执行节点查找操作，减少额外耗时。
3. 时间补偿机制：记录每次循环的开始时间，计算已消耗的时间，调整asyncio.sleep()的时长，确保每次循环总时长接近1秒。
4. 封装水箱状态：用类代替全局变量，维护水箱的状态，代码更清晰。
5. 统一异步循环：所有逻辑都放在一个异步主函数中，利用asyncio的调度机制实现精准定时。

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