4. 单阈值监控实现：用Python模拟真实场景

阈值监控，说白了就是给传感器数据画一条红线。

数据超过这条线，系统就要做出反应。我在做工业设备监控时，最常用的就是这种单阈值模式。简单、直接、可靠。今天我们就用Python和PyQt5，把这个过程完整走一遍。

4.1 单阈值监控的核心逻辑

先理清思路。单阈值监控就三个步骤：

• 采集数据：从传感器读取数值
• 比较判断：数值是否超过预设阈值
• 触发动作：报警、记录、或者两者同时

嗯，这里要注意。阈值本身可以是上限（超过就报警），也可以是下限（低于就报警）。我习惯把这两种情况统一处理，代码里加个方向参数就行。

4.2 Python模拟传感器数据

真实项目中，传感器数据往往有噪声。我一般用正弦波叠加随机噪声来模拟。这样既能看到趋势变化，又能测试阈值触发的稳定性。

import random
import math
import time

def simulate_sensor(base_value=50, amplitude=20, noise=5):
    """
    模拟传感器数据生成器
    base_value: 基准值
    amplitude: 波动幅度
    noise: 随机噪声
    """
    t = 0
    while True:
        # 正弦波 + 随机噪声
        value = base_value + amplitude * math.sin(t * 0.1) + random.uniform(-noise, noise)
        yield round(value, 2)
        t += 1
        time.sleep(0.5)  # 每0.5秒生成一个数据

为什么用生成器？因为在实际监控中，数据是源源不断的。生成器可以按需产出数据，不会一次性占用大量内存。我在做嵌入式数据采集时，也常用这种模式。

4.3 阈值监控引擎实现

监控引擎是整个系统的核心。它负责接收数据、判断阈值、触发动作。我建议把引擎设计成可配置的，这样后期维护起来方便。

class ThresholdMonitor:
    def __init__(self, threshold, direction="upper"):
        """
        threshold: 阈值
        direction: "upper" 上限 或 "lower" 下限
        """
        self.threshold = threshold
        self.direction = direction
        self.alert_callbacks = []  # 报警回调函数列表
        self.log_callbacks = []    # 记录回调函数列表
        self.last_value = None
        self.alert_active = False  # 防止重复报警

    def check(self, value):
        """检查当前值是否触发阈值"""
        if self.direction == "upper":
            triggered = value > self.threshold
        else:
            triggered = value < self.threshold

        # 触发且之前未报警，才执行动作
        if triggered and not self.alert_active:
            self.alert_active = True
            self._trigger_alert(value)
        elif not triggered:
            self.alert_active = False

        self.last_value = value
        return triggered

    def _trigger_alert(self, value):
        """触发报警和记录"""
        for cb in self.alert_callbacks:
            cb(value, self.threshold)
        for cb in self.log_callbacks:
            cb(value, self.threshold)

    def add_alert_callback(self, callback):
        self.alert_callbacks.append(callback)

    def add_log_callback(self, callback):
        self.log_callbacks.append(callback)

4.4 基于PyQt5的阈值显示界面

光有后台逻辑还不够，得让用户看到数据变化。PyQt5做这种小工具特别顺手。我一般用QWidget + QTimer实现实时刷新。

import sys
from PyQt5.QtWidgets import *
from PyQt5.QtCore import QTimer

class ThresholdUI(QWidget):
    def __init__(self, monitor):
        super().__init__()
        self.monitor = monitor
        self.sensor = simulate_sensor()
        self.init_ui()

    def init_ui(self):
        self.setWindowTitle("单阈值监控 - 实战演示")
        self.setGeometry(300, 300, 400, 300)

        # 布局
        layout = QVBoxLayout()

        # 当前值显示
        self.value_label = QLabel("当前值: --")
        self.value_label.setStyleSheet("font-size: 24px; font-weight: bold;")
        layout.addWidget(self.value_label)

        # 阈值显示
        self.threshold_label = QLabel(f"阈值: {self.monitor.threshold}")
        layout.addWidget(self.threshold_label)

        # 状态显示
        self.status_label = QLabel("状态: 正常")
        self.status_label.setStyleSheet("color: green; font-size: 18px;")
        layout.addWidget(self.status_label)

        # 日志区域
        self.log_text = QTextEdit()
        self.log_text.setReadOnly(True)
        layout.addWidget(self.log_text)

        self.setLayout(layout)

        # 定时器，每500ms更新一次
        self.timer = QTimer()
        self.timer.timeout.connect(self.update_data)
        self.timer.start(500)

    def update_data(self):
        value = next(self.sensor)
        self.value_label.setText(f"当前值: {value}")

        triggered = self.monitor.check(value)
        if triggered:
            self.status_label.setText("⚠️ 报警！超过阈值！")
            self.status_label.setStyleSheet("color: red; font-size: 18px;")
        else:
            self.status_label.setText("状态: 正常")
            self.status_label.setStyleSheet("color: green; font-size: 18px;")

    def on_alert(self, value, threshold):
        """报警回调"""
        self.log_text.append(f"[报警] 值 {value} 超过阈值 {threshold}")

    def on_log(self, value, threshold):
        """记录回调"""
        self.log_text.append(f"[记录] 值 {value}, 阈值 {threshold}")

4.5 阈值触发动作：报警与记录

触发动作是监控系统的最终目的。我把它分为两类：

动作类型	实现方式	典型场景
报警	界面变色、弹窗、声音	温度过高、压力异常
记录	写入日志、数据库	数据审计、故障分析

在实际项目中，报警和记录往往是同时进行的。报警让操作员立即知道异常，记录则用于事后追溯。我习惯把回调函数设计成可插拔的，这样想加什么动作都方便。

# 启动示例
if __name__ == "__main__":
    # 创建监控器，阈值设为60，上限模式
    monitor = ThresholdMonitor(threshold=60, direction="upper")

    # 创建UI
    app = QApplication(sys.argv)
    ui = ThresholdUI(monitor)

    # 注册回调
    monitor.add_alert_callback(ui.on_alert)
    monitor.add_log_callback(ui.on_log)

    ui.show()
    sys.exit(app.exec_())

4.6 知识体系总览

下面这张图，把单阈值监控的完整流程画出来了。你想想看，从传感器数据到最终动作，每一步都很清晰。

4.7 完整代码整合

把上面几段代码拼起来，就是一个可运行的单阈值监控系统。我个人习惯把代码分成三个文件：

• sensor.py - 传感器模拟
• monitor.py - 阈值监控引擎
• ui.py - PyQt5界面

这样结构清晰，后期改哪个模块都不影响其他部分。你在实际项目中也可以这么组织。

好了，单阈值监控的核心内容就这些。说白了就是三步：拿数据、比大小、做反应。下一节我们会在这个基础上，引入多阈值和动态阈值管理，让监控系统更智能。

---

公众号:蓝海资料掘金营，微信deep3321