4. 网关多路Modbus设备轮询调度器

任务目标

实现一个多路 Modbus 轮询调度器：为每个从站配置协议（RTU/TCP）、地址、周期，调度器按各自周期触发采集，采集结果写入统一数据点表。核心是周期调度 + 多路复用，不依赖真实硬件即可验证调度逻辑。

依赖与环境

• Python 3.8+，安装 pymodbus（真实采集用）与 pytest（可选）。
• 本案例用可注入的 read_func 抽象真实 IO，便于离线验证调度正确性。

调度器设计

每个设备是一个 DeviceTask，含周期 period 与下次到期时间 next_due。调度循环每次取最早到期的任务执行，执行后按周期推进 next_due。这样不同周期的设备互不阻塞，实现多路轮询。

完整实现

import time
import heapq
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional

@dataclass
class DeviceTask:
    name: str
    protocol: str          # "rtu" | "tcp"
    slave_id: int
    period: float          # 秒
    address: int = 0
    count: int = 1
    next_due: float = 0.0
    last_value: Optional[list] = None
    last_ts: float = 0.0

class PollScheduler:
    """多路 Modbus 轮询调度器：按各设备周期触发采集。"""

    def __init__(self, read_func: Callable[[DeviceTask], list]):
        self._read = read_func
        self._tasks: Dict[str, DeviceTask] = {}
        self._heap: List = []
        self.latest: Dict[str, dict] = {}

    def add_device(self, task: DeviceTask, now: float) -> None:
        task.next_due = now + task.period
        self._tasks[task.name] = task
        heapq.heappush(self._heap, (task.next_due, task.name))

    def _run_one(self, task: DeviceTask, now: float) -> None:
        try:
            values = self._read(task)
            task.last_value = values
            task.last_ts = now
            self.latest[task.name] = {
                "protocol": task.protocol,
                "slave_id": task.slave_id,
                "values": values,
                "ts": now,
                "ok": True,
            }
        except Exception as exc:  # 单路失败不影响其他设备
            self.latest[task.name] = {
                "protocol": task.protocol,
                "slave_id": task.slave_id,
                "values": None,
                "ts": now,
                "ok": False,
                "error": str(exc),
            }
        task.next_due = now + task.period
        heapq.heappush(self._heap, (task.next_due, task.name))

    def poll_once(self, now: float) -> Optional[str]:
        """执行一个到期任务，返回设备名；无到期任务返回 None。"""
        if not self._heap or self._heap[0][0] > now:
            return None
        _, name = heapq.heappop(self._heap)
        task = self._tasks[name]
        self._run_one(task, now)
        return name

    def run(self, duration: float, clock: Callable[[], float] = time.monotonic,
            sleep: Callable[[float], None] = time.sleep) -> None:
        start = clock()
        while clock() - start < duration:
            now = clock()
            if self.poll_once(now) is None:
                sleep(0.005)

# ---- 离线验证：用假时钟与假读取函数驱动调度 ----
if __name__ == "__main__":
    calls = []

    def fake_read(task: DeviceTask) -> list:
        calls.append((task.name, task.protocol, task.slave_id))
        return [task.slave_id * 10 + task.address]

    sched = PollScheduler(fake_read)
    sched.add_device(DeviceTask("rtu1", "rtu", 1, 0.1, address=0), now=0.0)
    sched.add_device(DeviceTask("tcp2", "tcp", 2, 0.2, address=5), now=0.0)
    sched.add_device(DeviceTask("rtu3", "rtu", 3, 0.5, address=2), now=0.0)

    # 模拟 0.5s 时间推进，步长 0.05s
    t = 0.0
    while t <= 0.5 + 1e-9:
        while sched.poll_once(t) is not None:
            pass
        t = round(t + 0.05, 3)

    for name in ("rtu1", "tcp2", "rtu3"):
        print(name, sched.latest[name]["values"], sched.latest[name]["ok"])
    print("total_calls", len(calls))

运行与验证

python scheduler.py

判定标准：

• 三个设备均出现在 latest 中且 ok=True。
• 周期 0.1s 的 rtu1 在 0.5s 内被调用次数最多，0.5s 的 rtu3 最少，体现按周期轮询。
• 把某个 fake_read 改为抛异常，该设备 ok=False 且带 error，其余设备仍正常采集，验证单路故障隔离。

接入真实 Modbus

把 read_func 换成真实读取即可：RTU 用 pymodbus.client.ModbusSerialClient，TCP 用 ModbusTcpClient，按 task.protocol 选择客户端，调用 read_holding_registers(task.address, task.count, slave=task.slave_id) 并返回 resp.registers。调度器本身无需改动。