多回路控制器分时调度仿真 —— 基于时间片轮转与工业过程控制

 

“那年调试一条间歇反应生产线，系统里挂着液位、压力、温度三条控制回路。PLC扫描周期太短，三个PID同时狂算，CPU负载直接飙红，模拟量输出还互相打架，导致调节阀振荡。后来我在上位机里重构了调度逻辑，引入时间片轮转（Time-Slicing），让液位、压力、温度三条回路分时循环运行。结果CPU占用率降了下来，各回路互不干扰，控制效果反而更稳定了。”

—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸

 

一、实际应用场景描述

 

在间歇化工、制药发酵、食品加工等批次生产（Batch Production）场景中，一套装置往往需要同时控制多个工艺参数。例如一个间歇反应釜：

 

                    ┌──────────────────────────────────────────────┐

                    │ 多回路控制器分时调度系统 │

                    │ │

                    │ [上位机 SCADA / DCS 系统] │

                    │ │ 调度指令 / 状态反馈 │

                    │ ▼ │

                    │ ┌────────────────────────────┐ │

                    │ │ 调度器 (Dispatcher) │ │

                    │ │ ┌──────────────────────┐ │ │

                    │ │ │ 1. 时间片管理 │ │ │

                    │ │ │ (Time Slicing) │ │ │

                    │ │ └──────────────────────┘ │ │

                    │ │ ┌──────────────────────┐ │ │

                    │ │ │ 2. 优先级仲裁 │ │ │

                    │ │ │ (Priority Arb.) │ │ │

                    │ │ └──────────────────────┘ │ │

                    │ └────────────┬───────────────┘ │

                    │ │ 分时使能信号 │

                    │ ┌───────┼───────┬───────┐ │

                    │ ▼ ▼ ▼ ▼ │

                    │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │

                    │ │ 液位回路 │ │ 压力回路 │ │ 温度回路 │ │

                    │ │ LC (LIC) │ │ PC (PIC) │ │ TC (TIC) │ │

                    │ └────┬────┘ └────┬────┘ └────┬────┘ │

                    │ │ │ │ │

                    │ ▼ ▼ ▼ │

                    │ ┌────────────────────────────────────┐ │

                    │ │ 模拟量输出模块 (AO) │ │

                    │ │ ┌───┐ ┌───┐ ┌───┐ │ │

                    │ │ │LV ├──►PV ├──►TV ├──... │ ← 互斥 │

                    │ │ └─┬─┘ └─┬─┘ └─┬─┘ │ │

                    │ │ └──────┴──────┘ │ │

                    │ └────────────┬─────────────────────┘ │

                    │ │ 4-20mA / 0-10V │

                    │ ┌───────┴───────┐ │

                    │ ▼ ▼ │

                    │ ┌─────────┐ ┌─────────┐ │

                    │ │ 调节阀LV │ │ 反应釜 │ │

                    │ │ (液位) │ │ (Process)│ │

                    │ └─────────┘ └────┬────┘ │

                    │ │ 工艺耦合 │

                    │ ┌───────────────┼───────────────┐ │

                    │ ▼ ▼ ▼ │

                    │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │

                    │ │ 压力变送器 │ │ 温度变送器 │ │ 液位变送器 │ │

                    │ │ (PT) │ │ (TT) │ │ (LT) │ │

                    │ └─────────┘ └─────────┘ └─────────┘ │

                    │ │ │ │ │

                    │ └───────────────┴───────────────┘ │

                    │ │ 4-20mA 反馈 │

                    │ ▼ │

                    │ ┌────────────────────────────────────┐ │

                    │ │ 模拟量输入模块 (AI) │ │

                    │ └────────────────────────────────────┘ │

                    │ │

                    │ 核心: 时间片轮转 + 资源互斥 + 解耦控制 │

                    └──────────────────────────────────────────────┘

 

并行计算 vs 分时调度

 

维度 并行计算（同时运行） 分时调度（轮流运行）

CPU负载 ❌ 峰值高，易过载 ✅ 平稳，可控

输出冲突 ❌ 调节阀振荡 ✅ 互斥，稳定

调节品质 ❌ 易互相干扰 ✅ 解耦，清晰

调试难度 ❌ 难以定位 ✅ 单回路排查

适用场景 连续流程，DCS 间歇过程，嵌入式

 

二、引入痛点

 

2.1 现场的真实困境

 

场景 现场发生了什么 根因

“CPU红灯狂闪” “PLC扫描超时” 多回路同时计算

“阀门来回抖” “液位还没稳，压力又动了” 输出资源竞争

“参数整定难” “调好液位，温度又乱了” 回路强耦合

“批次一致性差” “每批产品指标波动大” 控制时序混乱

“能耗异常” “蒸汽/冷却水浪费” 控制动作重叠

 

2.2 核心矛盾

 

工业PC或低端PLC的计算资源是有限的，而多个PID回路如果同时抢占CPU和输出资源，必然导致系统震荡甚至崩溃。 解决方案不是“更强的硬件”，而是“更合理的调度”。我们需要引入操作系统的时间片轮转思想，让多个控制回路分时复用CPU和模拟量输出资源，在保证控制周期足够的前提下，实现系统的稳定运行。

2.3 我们要解决什么

 

用一段精简的 Python 程序，构建一个多回路控制器分时调度仿真系统，实现：

 

1. 多回路建模 —— 液位、压力、温度三个典型回路

2. 时间片调度 —— 按固定时间片轮流激活回路

3. 优先级管理 —— 液位 > 压力 > 温度的调度权重

4. 资源互斥 —— 同一时刻只有一个回路输出

5. 可视化 —— 展示各回路PV、OP及调度时序

 

三、核心逻辑讲解

 

3.1 理论基础：时间片轮转（Round Robin）

 

本工具基于哈工程《工业过程控制》第一章“自动控制系统概述”和计算机控制技术：

 

① 调度周期（Schedule Period）

 

设系统主时钟周期为 T_{scan} ，每个回路分配一个时间片 T_{slice} 。

 

若有 N 个回路，则：

 

T_{schedule} = N \times T_{slice}

 

② 回路激活逻辑

 

在每个 T_{scan} 内，调度器判断当前时间所属的时间片，激活对应的回路：

 

if current_time % T_schedule in [0, T_slice):

    Activate(Loop_Level)

elif current_time % T_schedule in [T_slice, 2*T_slice):

    Activate(Loop_Pressure)

else:

    Activate(Loop_Temperature)

 

③ 输出保持（Hold）

 

未被激活的回路，其PID输出保持上一周期的值（或使用Zero-Order Hold），直到再次被调度。

 

3.2 增量式PID算法

 

为了适应分时调度，我们采用增量式PID，避免积分项在失活期间累积：

 

\Delta u(k) = K_p[e(k)-e(k-1)] + K_i e(k) + K_d[e(k)-2e(k-1)+e(k-2)]

 

u(k) = u(k-1) + \Delta u(k)

 

这样，当回路未被调度时， u(k) 保持不变。

 

四、代码讲解（面向对象设计）

 

4.1 类结构总览

 

类名 职责 设计模式

 

"ControlLoopType" 回路类型（Enum） 枚举

 

"LoopPriority" 优先级（Enum） 枚举

 

"ProcessVariable" 过程变量（dataclass） 值对象

 

"ControlSignal" 控制信号（dataclass） 值对象

 

"PIDController" PID控制器 策略模式

 

"ProcessModel" 被控过程模型 封装

 

"LevelProcess" 液位过程模型 继承

 

"PressureProcess" 压力过程模型 继承

 

"TemperatureProcess" 温度过程模型 继承

 

"ControlLoop" 控制回路（聚合） 组合

 

"TimeSliceScheduler" 时间片调度器 模板方法

 

"MultiLoopSystem" 多回路系统（聚合根） 聚合根

 

"VisualizationEngine" 可视化引擎 封装

 

4.2 数据模型与枚举

 

from dataclasses import dataclass, field

from typing import List, Dict, Optional, Tuple, Deque

from enum import Enum, auto

from collections import deque

import numpy as np

import matplotlib.pyplot as plt

from datetime import datetime

import math

 

class ControlLoopType(Enum):

    """控制回路类型"""

    LEVEL = auto()

    PRESSURE = auto()

    TEMPERATURE = auto()

 

class LoopPriority(Enum):

    """回路优先级（数值越小优先级越高）"""

    CRITICAL = 1 # 液位（防溢出/抽空）

    HIGH = 2 # 压力（防超压）

    MEDIUM = 3 # 温度（过程质量）

 

@dataclass(frozen=True)

class ProcessVariable:

    """过程变量 —— 值对象"""

    value: float

    unit: str

    timestamp: float = field(default_factory=lambda: datetime.now().timestamp())

    quality: str = "GOOD" # GOOD, BAD, UNCERTAIN

 

@dataclass

class ControlSignal:

    """控制信号 —— 值对象"""

    value: float # 0-100%

    loop_type: ControlLoopType

    is_active: bool = False

    timestamp: float = field(default_factory=lambda: datetime.now().timestamp())

 

4.3 PID控制器（策略模式）

 

class PIDController:

    """

    PID控制器 —— 策略模式

 

    采用增量式PID，适应分时调度

    """

 

    def __init__(self, kp: float = 1.0, ki: float = 0.1, kd: float = 0.05,

                 setpoint: float = 50.0, sample_time: float = 1.0):

        self.kp = kp

        self.ki = ki

        self.kd = kd

        self.setpoint = setpoint

        self.sample_time = sample_time

 

        # 状态变量

        self._prev_error = 0.0

        self._prev_prev_error = 0.0

        self._integral = 0.0

        self._last_output = 0.0

        self._last_time = datetime.now().timestamp()

 

        # 输出限幅

        self.output_limits = (0.0, 100.0) # 0-100%

 

    def compute(self, process_variable: float, dt: float) -> float:

        """

        计算PID增量输出

 

        Args:

            process_variable: 当前PV值

            dt: 实际经过的时间（用于积分和微分）

 

        Returns:

            控制增量 (0-100%)

        """

        current_time = datetime.now().timestamp()

 

        # 确保达到采样时间（即使被分时调度，也要保证最小间隔）

        if dt < self.sample_time * 0.5: # 允许一定误差

            return self._last_output

 

        # 计算误差

        error = self.setpoint - process_variable

 

        # 增量式PID计算

        # Δu = Kp*(e(k)-e(k-1)) + Ki*e(k)*dt + Kd*(e(k)-2e(k-1)+e(k-2))/dt

        p_term = self.kp * (error - self._prev_error)

 

        # 积分项（只在激活时累积，避免失活时积分暴走）

        self._integral += error * dt

        i_term = self.ki * self._integral

 

        # 微分项（注意：微分作用于PV而非误差，防止设定值突变引起冲击）

        if dt > 0:

            d_term = self.kd * (-(process_variable - (self.setpoint - self._prev_error)) / dt)

        else:

            d_term = 0.0

 

        # 计算增量

        delta_output = p_term + i_term + d_term

 

        # 更新输出

        output = self._last_output + delta_output

 

        # 限幅

        output = max(self.output_limits[0], min(output, self.output_limits[1]))

 

        # 抗积分饱和（反向计算）

        if output >= self.output_limits[1] and delta_output > 0:

            self._integral -= error * dt # 回退积分

        elif output <= self.output_limits[0] and delta_output < 0:

            self._integral -= error * dt # 回退积分

 

        # 更新状态

        self._prev_prev_error = self._prev_error

        self._prev_error = error

        self._last_output = output

        self._last_time = current_time

 

        return output

 

    def reset(self):

        """复位控制器"""

        self._prev_error = 0.0

        self._prev_prev_error = 0.0

        self._integral = 0.0

        self._last_output = 0.0

 

    def get_debug_info(self) -> Dict:

        return {

            "setpoint": self.setpoint,

            "pv": self.setpoint - self._prev_error,

            "error": self._prev_error,

            "kp": self.kp,

            "ki": self.ki,

            "kd": self.kd,

            "integral": self._integral,

            "output": self._last_output

        }

 

4.4 被控过程模型（基类与派生类）

 

class ProcessModel:

    """

    被控过程模型 —— 基类

 

    模拟被控对象的动态响应

    """

 

    def __init__(self, initial_value: float = 0.0, time_constant: float = 10.0,

                 gain: float = 1.0, dead_time: float = 0.0):

        self.state = initial_value

        self.time_constant = time_constant # 惯性时间常数 (s)

        self.gain = gain # 过程增益

        self.dead_time = dead_time # 纯滞后 (s)

        self.dead_time_buffer: Deque[float] = deque(maxlen=int(dead_time * 10) + 1)

 

    def update(self, control_input: float, dt: float) -> float:

        """

        更新过程状态

 

        Args:

            control_input: 控制输入 (0-100%)

            dt: 时间步长

 

        Returns:

            当前过程变量值

        """

        # 模拟纯滞后

        self.dead_time_buffer.append(control_input)

        if len(self.dead_time_buffer) > 0:

            delayed_input = self.dead_time_buffer[0]

        else:

            delayed_input = control_input

 

        # 一阶惯性环节: T * dy/dt + y = K * u

        # 离散化: y(k) = y(k-1) + (dt/T) * (K*u - y(k-1))

        self.state += (dt / self.time_constant) * (self.gain * delayed_input - self.state)

 

        # 加入过程噪声

        noise = np.random.normal(0, 0.1)

        return self.state + noise

 

    def get_state(self) -> float:

        return self.state

 

 

class LevelProcess(ProcessModel):

    """液位过程模型 —— 一阶惯性+积分特性"""

 

    def __init__(self, initial_level: float = 30.0):

        super().__init__(

            initial_value=initial_level,

            time_constant=15.0, # 液位变化较慢

            gain=0.8,

            dead_time=1.0

        )

        self.tank_height = 100.0 # cm

        self.outlet_valve_position = 30.0 # 出口阀开度 (%)

 

 

class PressureProcess(ProcessModel):

    """压力过程模型 —— 近似一阶惯性"""

 

    def __init__(self, initial_pressure: float = 101.3):

        super().__init__(

            initial_value=initial_pressure,

            time_constant=5.0, # 压力变化较快

            gain=0.5,

            dead_time=0.5

        )

 

 

class TemperatureProcess(ProcessModel):

    """温度过程模型 —— 大滞后，大惯性"""

 

    def __init__(self, initial_temp: float = 25.0):

        super().__init__(

            initial_value=initial_temp,

            time_constant=60.0, # 温度变化很慢

            gain=0.3,

            dead_time=10.0 # 大纯滞后

        )

 

4.5 控制回路（聚合）

 

class ControlLoop:

    """

    控制回路 —— 聚合根

 

    包含一个PID控制器和一个被控过程模型

    """

 

    def __init__(self, loop_type: ControlLoopType, priority: LoopPriority,

                 controller: PIDController, process: ProcessModel):

        self.loop_type = loop_type

        self.priority = priority

        self.controller = controller

        self.process = process

 

        self.is_enabled = True # 回路使能

        self.is_active = False # 当前是否被调度激活

        self.last_activation_time = 0.0

 

        # 历史数据

        self.pv_history: List[Tuple[float, float]] = []

        self.op_history: List[Tuple[float, float]] = []

        self.sp_history: List[Tuple[float, float]] = []

 

    def update(self, dt: float, current_time: float) -> ControlSignal:

        """

        更新回路状态

 

        Args:

            dt: 时间步长

            current_time: 当前时间

 

        Returns:

            控制信号

        """

        if not self.is_enabled:

            return ControlSignal(0.0, self.loop_type, False, current_time)

 

        # 获取当前PV

        pv = self.process.get_state()

        self.pv_history.append((current_time, pv))

        self.sp_history.append((current_time, self.controller.setpoint))

 

        # 只有被激活时才计算PID

        if self.is_active:

            op = self.controller.compute(pv, dt)

            self.last_activation_time = current_time

        else:

            # 未激活时保持上一输出

            op = self.controller._last_output

 

        # 更新过程模型

        self.process.update(op, dt)

        self.op_history.append((current_time, op))

 

        return ControlSignal(op, self.loop_type, self.is_active, current_time)

 

    def activate(self, current_time: float):

        """激活回路"""

        self.is_active = True

        self.last_activation_time = current_time

 

    def deactivate(self):

        """失活回路"""

        self.is_active = False

 

    def set_setpoint(self, sp: float):

        """设置设定值"""

        self.controller.setpoint = sp

 

    def get_status(self) -> Dict:

        debug = self.controller.get_debug_info()

        return {

            "loop_type": self.loop_type.name,

            "priority": self.priority.name,

            "enabled": self.is_enabled,

            "active": self.is_active,

            "pv": debug["pv"],

            "sp": debug["setpoint"],

            "op": debug["output"],

            "error": debug["error"]

        }

 

4.6 时间片调度器（模板方法）

 

class TimeSliceScheduler:

    """

    时间片调度器 —— 模板方法模式

 

    按固定时间片轮流激活控制回路

    """

 

    def __init__(self, time_slice: float = 2.0, scan_period: float = 0.1):

        self.time_slice = time_slice # 每个回路的时间片 (s)

        self.scan_period = scan_period # 系统扫描周期 (s)

        self.loops: List[ControlLoop] = []

        self.current_loop_index = 0

        self.last_switch_time = 0.0

        self.schedule_start_time = 0.0

 

        # 调度统计

        self.switch_count = 0

        self.execution_times: Dict[ControlLoopType, List[float]] = {}

 

    def add_loop(self, loop: ControlLoop):

        """添加控制回路（按优先级排序）"""

        self.loops.append(loop)

        self.loops.sort(key=lambda x: x.priority.value)

        self.execution_times[loop.loop_type] = []

 

    def schedule(self, current_time: float) -> Optional[ControlLoop]:

        """

        执行调度逻辑

 

        Returns:

            当前激活的回路（如果有）

        """

        if not self.loops:

            return None

 

        # 初始化调度开始时间

        if self.schedule_start_time == 0.0:

            self.schedule_start_time = current_time

            self.last_switch_time = current_time

 

        # 计算当前时间在调度周期中的位置

        elapsed_in_schedule = current_time - self.schedule_start_time

        total_schedule_period = self.time_slice * len(self.loops)

 

        # 判断是否到了切换时间

        if current_time - self.last_switch_time >= self.time_slice:

            # 切换到下一个回路

            self.current_loop_index = (self.current_loop_index + 1) % len(self.loops)

            self.last_switch_time = current_time

            self.switch_count += 1

 

            # 记录执行时间

            active_loop = self.loops[self.current_loop_index]

            self.execution_times[active_loop.loop_type].append(current_time)

 

        # 失活所有回路

        for loop in self.loops:

            loop.deactivate()

 

        # 激活当前回路

        if self.loops:

            active_loop = self.loops[self.current_loop_index]

            active_loop.activate(current_time)

            return active_loop

 

        return None

 

    def get_schedule_info(self) -> Dict:

        """获取调度信息"""

        return {

            "time_slice": self.time_slice,

            "scan_period": self.scan_period,

            "total_period": self.time_slice * len(self.loops),

            "current_index": self.current_loop_index,

            "switch_count": self.switch_count,

            "loop_count": len(self.loops)

        }

 

    def visualize_schedule_timeline(self, duration: float = 60.0) -> str:

        """生成调度时间线文本"""

        timeline = []

        timeline.append("调度时间线 (前60秒):")

        timeline.append("-" * 60)

 

        for loop_type, times in self.execution_times.items():

            line = f"{loop_type.name:12s}: "

            markers = []

            for t in times:

                if t <= duration:

                    markers.append(f"@{t:.1f}s")

            line += " ".join(markers[:10]) # 只显示前10个

            if len(markers) > 10:

                line += f"... (+{len(markers)-10} more)"

            timeline.append(line)

 

        return "\n".join(timeline)

 

4.7 多回路系统（聚合根）

 

class MultiLoopSystem:

    """

    多回路控制系统 —— 聚合根

 

    协调整个多回路控制流程

    """

 

    def __init__(self, time_slice: float = 2.0):

        # 初始化组件

        self.scheduler = TimeSliceScheduler(time_slice=time_slice, scan_period=0.1)

 

        # 创建回路（按优先级顺序添加）

        self._setup_control_loops()

 

        self.visualizer = VisualizationEngine()

        self.current_time = 0.0

        self.system_history: List[Tuple[float, Dict]] = []

 

    def _setup_control_loops(self):

        """配置控制回路"""

 

        # 1. 液位回路 (最高优先级)

        level_controller = PIDController(

            kp=1.5, ki=0.2, kd=0.1,

            setpoint=50.0, sample_time=2.0

        )

        level_process = LevelProcess(initial_level=30.0)

        level_loop = ControlLoop(

            ControlLoopType.LEVEL, LoopPriority.CRITICAL,

            level_controller, level_process

        )

        self.scheduler.add_loop(level_loop)

 

        # 2. 压力回路 (次高优先级)

        pressure_controller = PIDController(

            kp=2.0, ki=0.3, kd=0.05,

            setpoint=101.3, sample_time=2.0

        )

        pressure_process = PressureProcess(initial_pressure=101.3)

        pressure_loop = ControlLoop(

            ControlLoopType.PRESSURE, LoopPriority.HIGH,

            pressure_controller, pressure_process

        )

        self.scheduler.add_loop(pressure_loop)

 

        # 3. 温度回路 (中等优先级)

        temperature_controller = PIDController(

            kp=0.8, ki=0.05, kd=0.2,

            setpoint=80.0, sample_time=2.0

        )

        temperature_process = TemperatureProcess(initial_temp=25.0)

        temperature_loop = ControlLoop(

            ControlLoopType.TEMPERATURE, LoopPriority.MEDIUM,

            temperature_controller, temperature_process

        )

        self.scheduler.add_loop(temperature_loop)

 

    def run_simulation(self, duration: float = 120.0, dt: float = 0.1,

                       output: str = "multi_loop_control.png") -> str:

        """

        运行多回路控制仿真

        """

        print("=" * 60)

        print(" 多回路控制器分时调度仿真 v1.0")

        print(" 基于哈尔滨工程大学《工业过程控制》课程理论")

        print("=" * 60)

 

        schedule_info = self.scheduler.get_schedule_info()

        print(f"\n⚙️ 调度配置:")

        print(f" 时间片: {schedule_info['time_slice']}s")

        print(f" 扫描周期: {schedule_info['scan_period']}s")

        print(f" 调度周期: {schedule_info['total_period']}s")

        print(f" 回路数量: {schedule_info['loop_count']}")

 

        print(f"\n🚀 开始仿真...")

        steps = int(duration / dt)

 

        for i in range(steps):

            self.current_time = i * dt

 

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！