python的工业过程控制场景模拟第一百零四篇:巡检机器人数据同步程序，采集仪表参数实时上传过程控制系统数据库。
巡检机器人数据同步程序 —— 基于工业过程控制的数据采集与实时入库
“那年化工厂巡检，机器人明明拍到了压力表超限，上位机却半小时后才收到报警，差点酿成泄漏事故。后来我们用边缘缓存 + 断点续传 + 环形缓冲重构了采集链路，把数据延迟压到了 200ms 以内，再也没出现过‘看到了却来不及反应’的情况。”
—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸
一、实际应用场景描述
在石油化工、电力电站、制药车间等场景，巡检机器人需要7×24 小时对分散的仪表、阀门、设备进行数据采集，并实时同步到过程控制系统（PCS/DCS）：
┌──────────────────────────────────────────────┐
│ 巡检机器人数据同步系统 │
│ │
│ [巡检机器人本体] │
│ │ 视觉/红外/超声/振动 │
│ ▼ │
│ ┌────────────────────────────┐ │
│ │ 边缘采集层 │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 1. 仪表识别(OCR) │ │ │
│ │ │ (压力表/温度计) │ │ │
│ │ └──────────────────────┘ │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 2. 传感器融合 │ │ │
│ │ │ (振动+温度) │ │ │
│ │ └──────────────────────┘ │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 3. 本地缓存(环形缓冲) │ │ │
│ │ │ (掉线不丢数) │ │ │
│ └────────────┬───────────────┘ │
│ │ 结构化采样数据 │
│ ┌───────┴───────┐ │
│ ▼ ▼ │
│ ┌─────────┐ ┌─────────┐ │
│ │ 协议封装 │ │ 通信管理 │ │
│ │ • Modbus │ │ • 心跳检测 │ │
│ │ • MQTT │ │ • 重连机制 │ │
│ │ • OPC UA │ │ • QoS保障 │ │
│ └────┬────┘ └────┬────┘ │
│ │ 加密数据包 │ 链路状态 │
│ ▼ ▼ │
│ ┌────────────────────────────┐ │
│ │ 过程控制系统(PCS/DCS) │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 1. 实时数据库 │ │ │
│ │ │ (时序数据TSDB) │ │ │
│ │ └──────────────────────┘ │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 2. 报警服务 │ │ │
│ │ │ (阈值判断/联动) │ │ │
│ │ └──────────────────────┘ │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ 3. SCADA/HMI │ │ │
│ │ │ (操作员监控) │ │ │
│ └────────────┬───────────────┘ │
│ │ 控制指令/确认 │
│ ▼ │
│ ┌────────────────────────────┐ │
│ │ 物理世界 (高危环境) │ │
│ │ 🌡️ 压力表(0~10MPa) │ │
│ │ 💧 液位计(0~5m) │ │
│ │ ⚡ 振动传感器(0~50g) │ │
│ │ 🔥 红外热像(设备温度) │ │
│ └───────────────────────────┘ │
│ │
│ 核心: 实时采集 + 可靠传输 + 时序入库 + 报警联动 │
└──────────────────────────────────────────────┘
传统采集方式 vs 工业级同步方案
维度 传统采集（HTTP/串口轮询） 工业级数据同步
实时性 ❌ 秒级~分钟级延迟 ✅ 毫秒级 (<200ms)
可靠性 ❌ 断网即丢数 ✅ 边缘缓存 + 断点续传
数据完整性 ❌ 漏采、重复采 ✅ 序列号 + ACK 确认
并发能力 ❌ 单点阻塞 ✅ 异步采集 + 缓冲队列
报警及时性 ❌ 滞后严重 ✅ 边采边判 + 即时上报
工业兼容 ❌ 私有协议 ✅ Modbus/OPC UA/MQTT
二、引入痛点
2.1 现场的真实困境
场景 现场发生了什么 根因
“看到了却来不及” “压力超限30秒后才报警” 采集→上传链路过长
“断网全丢” “WiFi 闪断，半小时数据没了” 无边缘缓存
“数据打架” “同一个点位，SCADA 和机器人显示不一致” 时间戳/序列号缺失
“半夜误报” “凌晨3点误报压力异常，白跑一趟” 无滤波/死区处理
“系统撑不住” “100台机器人同时上传，数据库崩了” 无流量控制/背压
2.2 核心矛盾
工业数据的核心价值在于“实时性”和“确定性”。 巡检机器人不是摄像头，而是移动的过程控制前端。解决方案是：构建“采集—缓存—同步—入库—报警”的全链路闭环，用环形缓冲解决实时性，用 ACK 机制解决可靠性，用死区滤波解决有效性。
2.3 我们要解决什么
用一段精简的 Python 程序，构建一个 巡检机器人数据同步系统，实现：
1. 多源采集 —— 模拟仪表 OCR、传感器、设备状态
2. 边缘缓存 —— 环形缓冲 + 本地 SQLite，断网不丢数
3. 可靠传输 —— ACK 确认 + 重传机制
4. 实时入库 —— 时序数据库（模拟）批量写入
5. 即时报警 —— 阈值判断 + 边采边报
6. 可视化 —— 数据流向、延迟、完整性监控
三、核心逻辑讲解
3.1 理论基础：采样定理与数据同步
本工具基于哈工程《工业过程控制》第一章“过程控制系统概述”、第三章“信号检测与变送”和第九章“计算机控制系统”：
① 采样定理（香农定理）
为确保信号不失真，采样频率必须满足：
f_s \ge 2f_{max}
其中 f_{max} 为信号最高频率。对于工业过程：
- 压力/温度：1~10 Hz 足够
- 振动信号：1~5 kHz
- 本程序采用 5 Hz 采样率，兼顾实时性与带宽
② 数据同步模型
定义数据点结构：
DataPoint {
tag: str # 测点名称
timestamp: int # 纳秒级时间戳
value: float # 测量值
quality: int # 质量码 (0:好 1:可疑 2:坏)
seq: int # 序列号（防乱序）
}
③ 环形缓冲（Circular Buffer）
用于解决采集与上传速度不匹配问题：
- 写指针：采集线程写入
- 读指针：上传线程读取
- 当
"write - read == capacity" 时，触发背压（丢弃最旧或阻塞）
④ 死区滤波（Deadband Filter）
减少无效数据传输：
\text{transmit} =
\begin{cases}
true, & |v_{new} - v_{last}| > \Delta_{dead} \\
false, & \text{otherwise}
\end{cases}
3.2 系统架构总览
┌─────────────┐
│ 采集线程 │
│ (5Hz采样) │
└──────┬──────┘
│ DataPoint
┌─────────▼─────────┐
│ 环形缓冲 (512点) │
│ • 写指针++ │
│ • 读指针++ │
│ • 满则背压 │
└─────────┬─────────┘
│ 批量读取
┌─────────▼─────────┐
│ 上传管理器 │
│ • ACK确认 │
│ • 失败重传 │
│ • 心跳保活 │
└─────────┬─────────┘
│ 加密数据包
┌─────────▼─────────┐
│ 过程控制数据库 │
│ • 时序TSDB │
│ • 批量INSERT │
│ • 索引优化 │
└─────────┬─────────┘
│ 写入确认
┌─────────▼─────────┐
│ 报警服务 │
│ • 阈值判断 │
│ • 边采边报 │
│ • 联动输出 │
└───────────────────┘
四、代码讲解（面向对象设计）
4.1 类结构总览
类名 职责 设计模式
"DataPoint" 数据点（dataclass） 值对象
"SensorType" 传感器类型枚举 枚举
"CircularBuffer" 环形缓冲（线程安全） 生产者-消费者
"SensorSimulator" 传感器模拟器 工厂模式
"DatabaseConnector" 数据库连接（模拟） 适配器模式
"AlarmManager" 报警管理 观察者模式
"DataSynchronizer" 数据同步核心（聚合根） 聚合根
"VisualizationEngine" 可视化引擎 封装
4.2 核心代码（完整可运行）
完整源码约 520 行，包含 8 个类、多线程采集、环形缓冲、ACK 重传、报警联动。
以下为精简核心版，完整代码可直接复制运行。
<details><summary>🔧 完整源码（点击展开/折叠）</summary>
"""
巡检机器人数据同步程序 —— 工业级实时采集与入库
参考哈尔滨工程大学《工业过程控制》第三章信号检测与第九章计算机控制
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple, Any
from enum import Enum, auto
import time
import threading
import sqlite3
import json
import math
import random
from collections import deque
from datetime import datetime
import numpy as np
import matplotlib.pyplot as plt
from queue import Queue, Empty, Full
import logging
# ============================================================
# 1. 基础数据结构
# ============================================================
class SensorType(Enum):
"""传感器类型枚举"""
PRESSURE = auto() # 压力
TEMPERATURE = auto() # 温度
LEVEL = auto() # 液位
VIBRATION = auto() # 振动
INFRARED = auto() # 红外热像
@dataclass
class DataPoint:
"""工业数据点 —— 值对象（不可变）"""
tag: str # 测点名
timestamp: int # 纳秒时间戳
value: float # 测量值
quality: int = 0 # 质量码: 0=Good, 1=Uncertain, 2=Bad
seq: int = 0 # 序列号（防乱序）
unit: str = "" # 单位
deadband: float = 0.0 # 死区阈值
def to_dict(self) -> dict:
return {
'tag': self.tag,
'timestamp': self.timestamp,
'value': self.value,
'quality': self.quality,
'seq': self.seq,
'unit': self.unit
}
def to_json(self) -> str:
return json.dumps(self.to_dict())
class QualityCode:
"""质量码常量"""
GOOD = 0
UNCERTAIN = 1
BAD = 2
SENSOR_FAILURE = 3
COMM_FAILURE = 4
# ============================================================
# 2. 线程安全环形缓冲
# ============================================================
class CircularBuffer:
"""
线程安全环形缓冲 —— 生产者-消费者模式
用于解耦采集线程与上传线程
"""
def __init__(self, capacity: int = 512):
self.capacity = capacity
self.buffer = [None] * capacity
self.write_idx = 0
self.read_idx = 0
self.count = 0
self.lock = threading.RLock()
self.not_empty = threading.Condition(self.lock)
self.not_full = threading.Condition(self.lock)
def put(self, item: DataPoint, block: bool = True, timeout: float = 1.0) -> bool:
"""写入数据（生产者）"""
with self.not_full:
if self.count == self.capacity:
if not block:
return False
# 缓冲区满，触发背压
self.not_full.wait(timeout)
if self.count == self.capacity:
return False
self.buffer[self.write_idx] = item
self.write_idx = (self.write_idx + 1) % self.capacity
self.count += 1
self.not_empty.notify()
return True
def get(self, block: bool = True, timeout: float = 1.0) -> Optional[DataPoint]:
"""读取数据（消费者）"""
with self.not_empty:
if self.count == 0:
if not block:
return None
self.not_empty.wait(timeout)
if self.count == 0:
return None
item = self.buffer[self.read_idx]
self.buffer[self.read_idx] = None
self.read_idx = (self.read_idx + 1) % self.capacity
self.count -= 1
self.not_full.notify()
return item
def get_batch(self, batch_size: int = 10) -> List[DataPoint]:
"""批量读取"""
batch = []
for _ in range(batch_size):
item = self.get(block=False)
if item is None:
break
batch.append(item)
return batch
def size(self) -> int:
with self.lock:
return self.count
def is_full(self) -> bool:
with self.lock:
return self.count == self.capacity
def is_empty(self) -> bool:
with self.lock:
return self.count == 0
# ============================================================
# 3. 传感器模拟器（工厂模式）
# ============================================================
class SensorSimulator:
"""
传感器模拟器 —— 工厂模式
模拟各类工业传感器的真实行为（含噪声、漂移、故障）
"""
def __init__(self, sensor_id: str, sensor_type: SensorType,
base_value: float, noise_level: float = 0.02):
self.sensor_id = sensor_id
self.sensor_type = sensor_type
self.base_value = base_value
self.noise_level = noise_level
self.drift = 0.0
self.drift_rate = random.uniform(-0.001, 0.001)
self.failure_prob = 0.001 # 故障概率
self.last_value = base_value
self.units = {
SensorType.PRESSURE: "MPa",
SensorType.TEMPERATURE: "°C",
SensorType.LEVEL: "m",
SensorType.VIBRATION: "g",
SensorType.INFRARED: "°C"
}
def read(self) -> Tuple[float, int]:
"""读取传感器值（含噪声和故障模拟）"""
# 模拟故障
if random.random() < self.failure_prob:
return float('nan'), QualityCode.SENSOR_FAILURE
# 模拟漂移
self.drift += self.drift_rate
if abs(self.drift) > 0.1:
self.drift_rate *= -1
# 模拟噪声
noise = random.gauss(0, self.noise_level * self.base_value)
# 模拟过程动态（正弦波动）
process_variation = 0.05 * self.base_value * math.sin(time.time() * 0.1)
value = self.base_value + self.drift + noise + process_variation
self.last_value = value
# 合理性检查
if self.sensor_type == SensorType.PRESSURE and (value < 0 or value > 15):
return value, QualityCode.UNCERTAIN
if self.sensor_type == SensorType.TEMPERATURE and (value < -50 or value > 200):
return value, QualityCode.UNCERTAIN
return value, QualityCode.GOOD
def get_tag_name(self) -> str:
prefix = {
SensorType.PRESSURE: "PT",
SensorType.TEMPERATURE: "TE",
SensorType.LEVEL: "LT",
SensorType.VIBRATION: "VT",
SensorType.INFRARED: "IR"
}
return f"{prefix.get(self.sensor_type, 'AI')}_{self.sensor_id}"
def get_unit(self) -> str:
return self.units.get(self.sensor_type, "")
# ============================================================
# 4. 数据库连接（模拟时序数据库）
# ============================================================
class DatabaseConnector:
"""
数据库连接 —— 适配器模式
模拟工业时序数据库（如 InfluxDB、TimescaleDB）
"""
def __init__(self, db_path: str = ":memory:"):
self.db_path = db_path
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self.lock = threading.Lock()
self._init_db()
def _init_db(self):
with self.lock:
cursor = self.conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS process_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tag TEXT NOT NULL,
timestamp INTEGER NOT NULL,
value REAL NOT NULL,
quality INTEGER NOT NULL,
seq INTEGER NOT NULL,
unit TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_tag_time
ON process_data(tag, timestamp)
""")
self.conn.commit()
def insert_batch(self, points: List[DataPoint]) -> bool:
"""批量插入数据点"""
if not points:
return True
try:
with self.lock:
cursor = self.conn.cursor()
data = [
(p.tag, p.timestamp, p.value, p.quality, p.seq, p.unit)
for p in points
]
cursor.executemany("""
INSERT INTO process_data
(tag, timestamp, value, quality, seq, unit)
VALUES (?, ?, ?, ?, ?, ?)
""", data)
self.conn.commit()
return True
except Exception as e:
logging.error(f"数据库插入失败: {e}")
return False
def query_latest(self, tag: str, limit: int = 10) -> List[Dict]:
"""查询最新数据"""
with self.lock:
cursor = self.conn.cursor()
cursor.execute("""
SELECT tag, timestamp, value, quality, unit
FROM process_data
WHERE tag = ?
ORDER BY timestamp DESC
LIMIT ?
""", (tag, limit))
columns = [desc[0] for desc in cursor.description]
return [dict(zip(columns, row)) for row in cursor.fetchall()]
def get_statistics(self, tag: str, start_time: int, end_time: int) -> Dict:
"""获取统计数据"""
with self.lock:
cursor = self.conn.cursor()
cursor.execute("""
SELECT
COUNT(*) as count,
AVG(value) as avg_value,
MIN(value) as min_value,
MAX(value) as max_value,
SUM(CASE WHEN quality = 0 THEN 1 ELSE 0 END) as good_count
FROM process_data
WHERE tag = ? AND timestamp BETWEEN ? AND ?
""", (tag, start_time, end_time))
result = cursor.fetchone()
if result:
return {
'count': result[0],
'avg': result[1],
'min': result[2],
'max': result[3],
'good_rate': result[4] / result[0] if result[0] > 0 else 0
}
return {}
# ============================================================
# 5. 报警管理器（观察者模式）
# ============================================================
class AlarmManager:
"""
报警管理器 —— 观察者模式
支持阈值报警、变化率报警、质量报警
"""
def __init__(self):
self.alarms = {}
self.active_alarms = {}
self.callbacks = []
def add_threshold(self, tag: str, low: float = None, high: float = None):
"""添加阈值报警"""
self.alarms[tag] = {
'low': low,
'high': high,
'enabled': True
}
def check_alarm(self, point: DataPoint) -> Optional[Dict]:
"""检查报警条件"""
if point.quality != QualityCode.GOOD:
alarm = {
'tag': point.tag,
'type': 'QUALITY',
'severity': 'HIGH',
'message': f"质量异常: 质量码={point.quality}",
'value': point.value,
'timestamp': point.timestamp
}
return alarm
if point.tag not in self.alarms:
return None
config = self.alarms[point.tag]
alarm = None
if config['high'] is not None and point.value > config['high']:
alarm = {
'tag': point.tag,
'type': 'HIGH',
'severity': 'HIGH',
'message': f"超限报警: {point.value:.2f} > {config['high']}",
'value': point.value,
'threshold': config['high'],
'timestamp': point.timestamp
}
elif config['low'] is not None and point.value < config['low']:
alarm = {
'tag': point.tag,
'type': 'LOW',
'severity': 'MEDIUM',
'message': f"低限报警: {point.value:.2f} < {config['low']}",
'value': point.value,
'threshold': config['low'],
'timestamp': point.timestamp
}
if alarm:
self.active_alarms[point.tag] = alarm
self._notify_callbacks(alarm)
return alarm
# 报警恢复
if point.tag in self.active_alarms:
recovery = {
'tag': point.tag,
'type': 'RECOVERY',
'severity': 'INFO',
'message': f"报警恢复: 当前值={point.value:.2f}",
'value': point.value,
'timestamp': point.timestamp
}
del self.active_alarms[point.tag]
self._notify_callbacks(recovery)
return recovery
return None
def add_callback(self, callback):
"""添加报警回调"""
self.callbacks.append(callback)
def _notify_callbacks(self, alarm: Dict):
for cb in self.callbacks:
try:
cb(alarm)
except Exception as e:
logging.error(f"报警回调执行失败: {e}")
# ============================================================
# 6. 数据同步核心（聚合根）
# ============================================================
class DataSynchronizer:
"""
数据同步核心 —— 聚合根
协调采集、缓冲、上传、入库、报警全流程
"""
def __init__(self, buffer_capacity: int = 512):
self.buffer = CircularBuffer(buffer_capacity)
self.db = DatabaseConnector()
self.alarm_manager = AlarmManager()
self.sensors: Dict[str, SensorSimulator] = {}
self.running = False
self.seq_counter = 0
self.stats = {
'collected': 0,
'uploaded': 0,
'dropped': 0,
'alarms': 0,
'start_time': 0
}
# 死区配置
self.deadbands = {}
# 线程
self.collector_thread = None
self.uploader_thread = None
# 初始化传感器
self._init_sensors()
self._init_alarms()
def _init_sensors(self):
"""初始化传感器"""
# 压力变送器
self.add_sensor("PT101", SensorType.PRESSURE, 2.5, 0.01)
self.add_sensor("PT102", SensorType.PRESSURE, 1.8, 0.02)
# 温度变送器
self.add_sensor("TE201", SensorType.TEMPERATURE, 85.0, 0.5)
self.add_sensor("TE202", SensorType.TEMPERATURE, 120.0, 0.5)
# 液位计
self.add_sensor("LT301", SensorType.LEVEL, 2.8, 0.05)
# 振动传感器
self.add_sensor("VT401", SensorType.VIBRATION, 2.5, 0.1)
# 红外热像
self.add_sensor("IR501", SensorType.INFRARED, 65.0, 1.0)
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！