工艺管道智能巡检机器人仿真 —— 基于路径规划与定点采样的OOP实战

 

"去年在一个大型石化厂做技改，工艺员跟我吐槽：'这条管廊800米长，上下三层，每天要人工抄表4次，夏天钢管表面60℃，冬天寒风刺骨，年轻人根本不愿意干。'后来我们上了轨道式巡检机器人——沿着管廊预设磁轨行走，每到一块压力表、温度计下方自动停下，升降机构举升到表计高度，工业相机OCR识别读数，红外热像仪测温度，10分钟跑完一圈，数据直接入库。今天用程序把这个逻辑完整复现一遍。"

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

 

一、实际应用场景描述

 

在石油、化工、电力等行业，长距离工艺管廊上分布着大量的就地指示仪表（压力表、温度计、流量计）。传统的人工巡检存在劳动强度大、数据录入滞后、恶劣环境风险高等问题。智能巡检机器人通过预设路径导航，按既定路线遍历所有仪表点位，实现定时、自动、标准化的数据采集。

 

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

                    │ 工艺管道智能巡检系统 │

                    │ │

                    │ [调度中心 / 中控室] │

                    │ │ 任务下发/数据回传 │

                    │ ▼ │

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

                    │ │ 机器人控制站(上位机) │ │

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

                    │ │ │ 路径规划与任务调度 │ │ │

                    │ │ │ 仪表点位数据库 │ │ │

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

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

                    │ │ 无线通信(WiFi/4G) │

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

                    │ │ 巡检机器人本体 │ │

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

                    │ │ │ 移动底盘(轨道) │ │ │

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

                    │ │ ▼ │ │

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

                    │ │ │ 升降/旋转云台 │ │ │

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

                    │ │ ▼ │ │

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

                    │ │ │ 感知单元 │ │ │

                    │ │ │ • 可见光相机 │ │ │

                    │ │ │ • 红外热像仪 │ │ │

                    │ │ │ • 激光测距 │ │ │

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

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

                    │ │ │

                    │ ▼ │

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

                    │ │ 工艺管廊现场 │ │

                    │ │ ▓▓▓ ▓▓▓ ▓▓▓ ▓▓▓ │ ← 管道 │

                    │ │ [P1] [T2] [F3] [P4] │ ← 仪表点位 │

                    │ │ ↑ ↑ ↑ ↑ │ │

                    │ │ └─────轨道─────┘ │ │

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

                    │ │

                    │ 核心: 路径规划 + 定点停靠 + 自动采集 │

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

 

人工巡检 vs 机器人巡检

 

维度 人工巡检 机器人巡检

效率 2小时/圈 10分钟/圈

频次 4次/天 24次/天

数据准确性 手工记录易出错 自动采集100%准确

环境适应性 受限（高温/有毒） 全天候运行

成本 人力成本高 一次性投入，长期节省

数据价值 孤立数据 趋势分析+预测性维护

 

二、引入痛点

 

2.1 现场的真实困境

 

场景 现场发生了什么 根因

"抄表抄错了" "压力值抄成温度值" 人工疲劳/注意力分散

"漏检了" "半夜的巡检睡着了" 人性化管理难题

"表太高够不着" "爬梯危险还不准确" 仪表安装位置不合理

"数据滞后" "早上8点发现昨晚泄漏" 巡检频次不足

"恶劣天气" "暴雨大雪照样得出去" 人工巡检硬性要求

 

2.2 核心矛盾

 

工艺安全要求"高频次、全覆盖、高精度"巡检，但人力资源有限且不可靠。 智能巡检机器人的核心价值不在于"替代人"，而在于用标准化的自动化流程，解决"必须做但人不愿意/做不到"的任务。其技术核心是：如何在复杂管廊环境中，精确导航到每个仪表点位，并在正确的位置、正确的时间执行正确的采集动作。

2.3 我们要解决什么

 

用一段精简的 Python 程序，构建一个工艺管道巡检机器人仿真系统，实现：

 

1. 工艺管廊建模 —— 定义管道、仪表点位、障碍物

2. 路径规划 —— 生成最优巡检路线（顺序/往返）

3. 机器人运动模型 —— 速度、加速度、启停控制

4. 定点停靠逻辑 —— 到达仪表位置自动减速、精确定位

5. 仪表数据采集 —— 模拟温度、压力读数（含噪声）

6. 任务调度 —— 定时启动、循环执行

7. 数据记录与可视化 —— 轨迹、采集结果、统计报告

 

三、核心逻辑讲解

 

3.1 理论基础：移动机器人路径规划

 

本工具基于哈工程《工业过程控制》第十六章"计算机集成制造系统(CIMS)"和移动机器人控制相关内容：

 

① 点位模型

 

Waypoint = {

    'id': 'P101',

    'type': 'pressure',

    'position': (x, y, z), # 沿轨道距离、横向偏移、高度

    'normal_range': (min, max),

    'alarm_threshold': (low, high)

}

 

② 运动学模型（简化）

 

v(t) = v_{max} \cdot \tanh\left(\frac{t}{\tau}\right)

 

s(t) = \int_0^t v(\tau)d\tau

 

③ 定点停靠控制（PID位置环）

 

u(t) = K_p e(t) + K_i \int e(t)dt + K_d \frac{de(t)}{dt}

 

其中 e(t) = x_{target} - x_{current} 

 

④ 巡检任务调度

 

START → 移动到P1 → 采集数据 → 移动到P2 → 采集数据 → ... → END → LOOP

 

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

 

4.1 类结构总览

 

类名 职责 设计模式

 

"InstrumentSpec" 仪表规格（dataclass） 值对象

 

"Waypoint" 路径点（dataclass） 值对象

 

"RobotMotionModel" 机器人运动学模型 模板方法

 

"PIDPositionController" 位置PID控制器 封装

 

"SensorSimulator" 传感器数据模拟器 策略模式

 

"InspectionPlanner" 巡检路径规划器 策略模式

 

"DataRecorder" 数据记录与存储 封装

 

"VisualizationHelper" 轨迹与数据可视化 封装

 

"InspectionRobot" 巡检机器人本体 状态模式

 

"SimulationSystem" 系统编排器（聚合根） 聚合根

 

4.2 数据模型层

 

from dataclasses import dataclass, field

from datetime import datetime

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

import math

import random

 

@dataclass(frozen=True)

class InstrumentSpec:

    """仪表规格 —— 值对象"""

    id: str

    name: str

    type: Literal['temperature', 'pressure', 'flow'] # 仪表类型

    unit: str

    normal_min: float

    normal_max: float

    warning_low: Optional[float] = None

    warning_high: Optional[float] = None

    critical_low: Optional[float] = None

    critical_high: Optional[float] = None

    accuracy: float = 0.01 # 测量精度

 

@dataclass

class Waypoint:

    """路径点 —— 值对象"""

    id: str

    position: float # 沿轨道的一维坐标 (m)

    height_offset: float = 1.5 # 相对机器人底座的高度 (m)

    lateral_offset: float = 0.0 # 横向偏移 (m)

    instrument: Optional[InstrumentSpec] = None

    dwell_time: float = 3.0 # 停留采集时间 (s)

    approach_speed: float = 0.2 # 接近速度 (m/s)

    departure_speed: float = 0.5 # 离开速度 (m/s)

 

    def __lt__(self, other):

        return self.position < other.position

 

4.3 机器人运动模型

 

class RobotMotionModel:

    """

    巡检机器人运动学模型

 

    简化为一维轨道运动

    包含速度、加速度限制

    """

 

    def __init__(self, max_speed: float = 1.0, max_accel: float = 0.3,

                 start_pos: float = 0.0):

        self.max_speed = max_speed # 最大速度 (m/s)

        self.max_accel = max_accel # 最大加速度 (m/s²)

        self.position = start_pos # 当前位置 (m)

        self.velocity = 0.0 # 当前速度 (m/s)

        self.target_position = start_pos

        self._last_update_time = None

 

    def update(self, dt: float) -> Tuple[float, float]:

        """

        更新运动状态

 

        Args:

            dt: 步长 (s)

 

        Returns:

            (新位置, 新速度)

        """

        # 计算位置误差

        error = self.target_position - self.position

 

        # 梯形速度规划

        if abs(error) < 0.01: # 到达目标

            self.velocity = 0.0

            self.position = self.target_position

        else:

            # 计算达到目标所需的最小减速度

            stopping_dist = (self.velocity ** 2) / (2 * self.max_accel)

 

            if abs(error) <= stopping_dist and self.velocity > 0:

                # 需要减速

                accel = -self.max_accel

            elif abs(error) <= stopping_dist and self.velocity < 0:

                accel = self.max_accel

            else:

                # 可以加速或匀速

                direction = 1 if error > 0 else -1

                current_max_v = min(

                    self.max_speed,

                    math.sqrt(2 * self.max_accel * abs(error))

                )

                if abs(self.velocity) < current_max_v:

                    accel = direction * self.max_accel

                else:

                    accel = 0 # 匀速

 

            # 更新速度和位置

            new_velocity = self.velocity + accel * dt

            new_velocity = max(-self.max_speed, min(self.max_speed, new_velocity))

 

            self.position += (self.velocity + new_velocity) / 2 * dt

            self.velocity = new_velocity

 

        return self.position, self.velocity

 

    def set_target(self, target: float):

        """设置目标位置"""

        self.target_position = target

 

    def get_status(self) -> Dict:

        return {

            'position': round(self.position, 3),

            'velocity': round(self.velocity, 3),

            'target': round(self.target_position, 3)

        }

 

4.4 PID位置控制器

 

class PIDPositionController:

    """位置PID控制器（用于精确定位）"""

 

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

        self.kp = kp

        self.ki = ki

        self.kd = kd

        self.integral = 0.0

        self.prev_error = 0.0

 

    def update(self, target: float, current: float, dt: float) -> float:

        error = target - current

 

        # P

        p_term = self.kp * error

 

        # I

        self.integral += error * dt

        i_term = self.ki * self.integral

 

        # D

        derivative = (error - self.prev_error) / dt if dt > 0 else 0

        d_term = self.kd * derivative

 

        output = p_term + i_term + d_term

 

        # 限幅（输出为速度指令）

        output = max(-self.kp * 2, min(self.kp * 2, output))

 

        self.prev_error = error

        return output

 

4.5 传感器数据模拟器

 

class SensorSimulator:

    """传感器数据模拟器 —— 策略模式"""

 

    @staticmethod

    def simulate_temperature(spec: InstrumentSpec, base_temp: float = 85.0) -> float:

        """模拟温度读数"""

        # 基础值 + 随机漂移 + 噪声

        drift = random.uniform(-0.5, 0.5)

        noise = random.gauss(0, spec.accuracy * base_temp)

        value = base_temp + drift + noise

        return max(spec.normal_min * 0.9, min(spec.normal_max * 1.1, value))

 

    @staticmethod

    def simulate_pressure(spec: InstrumentSpec, base_pressure: float = 1.2) -> float:

        """模拟压力读数 (MPa)"""

        drift = random.uniform(-0.02, 0.02)

        noise = random.gauss(0, spec.accuracy * base_pressure)

        value = base_pressure + drift + noise

        return max(spec.normal_min * 0.9, min(spec.normal_max * 1.1, value))

 

    @staticmethod

    def simulate_flow(spec: InstrumentSpec, base_flow: float = 45.0) -> float:

        """模拟流量读数 (m³/h)"""

        drift = random.uniform(-0.5, 0.5)

        noise = random.gauss(0, spec.accuracy * base_flow)

        value = base_flow + drift + noise

        return max(spec.normal_min * 0.9, min(spec.normal_max * 1.1, value))

 

    def read_instrument(self, instrument: InstrumentSpec) -> Dict:

        """读取仪表数据"""

        if instrument.type == 'temperature':

            value = self.simulate_temperature(instrument)

        elif instrument.type == 'pressure':

            value = self.simulate_pressure(instrument)

        elif instrument.type == 'flow':

            value = self.simulate_flow(instrument)

        else:

            value = 0.0

 

        # 判断状态

        status = 'NORMAL'

        if instrument.critical_low and value <= instrument.critical_low:

            status = 'CRITICAL_LOW'

        elif instrument.critical_high and value >= instrument.critical_high:

            status = 'CRITICAL_HIGH'

        elif instrument.warning_low and value <= instrument.warning_low:

            status = 'WARNING_LOW'

        elif instrument.warning_high and value >= instrument.warning_high:

            status = 'WARNING_HIGH'

 

        return {

            'value': round(value, 3),

            'unit': instrument.unit,

            'status': status,

            'timestamp': datetime.now().isoformat()

        }

 

4.6 巡检路径规划器

 

class InspectionPlanner:

    """巡检路径规划器 —— 策略模式"""

 

    def __init__(self, waypoints: List[Waypoint]):

        self.waypoints = sorted(waypoints)

        self.total_length = self.waypoints[-1].position if waypoints else 0

 

    def plan_route(self, start_index: int = 0,

                   pattern: str = 'round_trip') -> List[Waypoint]:

        """

        规划巡检路线

 

        Args:

            start_index: 起始点位索引

            pattern: 'one_way'(单向), 'round_trip'(往返), 'patrol'(循环)

 

        Returns:

            按顺序访问的点位列表

        """

        if not self.waypoints:

            return []

 

        route = []

 

        if pattern == 'one_way':

            # 从起点到终点

            route = [wp for wp in self.waypoints if wp.position >= self.waypoints[start_index].position]

 

        elif pattern == 'round_trip':

            # 去程 + 返程

            forward = [wp for wp in self.waypoints if wp.position >= self.waypoints[start_index].position]

            backward = list(reversed([wp for wp in self.waypoints if wp.position < self.waypoints[start_index].position]))

            route = forward + backward

 

        elif pattern == 'patrol':

            # 循环巡检（从起点开始完整一圈）

            route = self.waypoints[start_index:] + self.waypoints[:start_index]

 

        return route

 

    def estimate_duration(self, robot: RobotMotionModel,

                         route: List[Waypoint]) -> float:

        """估算巡检时长"""

        total_time = 0.0

        current_pos = route[0].position if route else 0

 

        for wp in route:

            # 移动时间（简化：匀速）

            distance = abs(wp.position - current_pos)

            move_time = distance / robot.max_speed

            total_time += move_time

            total_time += wp.dwell_time # 停留采集时间

            current_pos = wp.position

 

        return total_time

 

4.7 数据记录器

 

class DataRecorder:

    """巡检数据记录器"""

 

    def __init__(self):

        self.records: List[Dict] = []

        self.session_id = datetime.now().strftime("%Y%m%d_%H%M%S")

 

    def record_reading(self, waypoint: Waypoint, reading: Dict):

        """记录仪表读数"""

        record = {

            'session_id': self.session_id,

            'waypoint_id': waypoint.id,

            'instrument_id': waypoint.instrument.id if waypoint.instrument else 'N/A',

            'position': waypoint.position,

            'timestamp': reading['timestamp'],

            'value': reading['value'],

            'unit': reading['unit'],

            'status': reading['status']

        }

        self.records.append(record)

 

    def get_summary(self) -> Dict:

        """获取巡检摘要"""

        if not self.records:

            return {}

 

        total = len(self.records)

        normal = sum(1 for r in self.records if r['status'] == 'NORMAL')

        warnings = sum(1 for r in self.records if 'WARNING' in r['status'])

        criticals = sum(1 for r in self.records if 'CRITICAL' in r['status'])

 

        return {

            'session_id': self.session_id,

            'total_readings': total,

            'normal': normal,

            'warnings': warnings,

            'criticals': criticals,

            'success_rate': normal / total * 100 if total > 0 else 0

        }

 

    def export_csv(self, filename: str = None):

        """导出CSV（简化版，仅打印）"""

        if not filename:

            filename = f"inspection_{self.session_id}.csv"

 

        print(f"\n📄 巡检数据 ({filename}):")

        print("点位ID, 仪表ID, 位置(m), 时间, 数值, 单位, 状态")

        for r in self.records:

            print(f"{r['waypoint_id']}, {r['instrument_id']}, {r['position']:.2f}, "

                  f"{r['timestamp']}, {r['value']}, {r['unit']}, {r['status']}")

 

4.8 可视化助手

 

class VisualizationHelper:

    """巡检可视化助手"""

 

    @staticmethod

    def plot_inspection(route: List[Waypoint], robot_trajectory: List[Tuple[float, float]],

                        readings: List[Dict], output: str = "inspection.png"):

        import matplotlib.pyplot as plt

 

        fig, axes = plt.subplots(2, 2, figsize=(14, 10))

 

        # 1. 管廊布局与点位

        ax = axes[0, 0]

        positions = [wp.position for wp in route]

        heights = [wp.height_offset for wp in route]

        colors = ['red' if wp.instrument and wp.instrument.type == 'temperature' else

                  'blue' if wp.instrument and wp.instrument.type == 'pressure' else

                  'green' for wp in route]

 

        ax.scatter(positions, heights, c=colors, s=100, alpha=0.7, edgecolors='black')

        for wp in route:

            ax.annotate(wp.id, (wp.position, wp.height_offset),

                       xytext=(5, 5), textcoords='offset points', fontsize=8)

 

        ax.plot([0, max(positions)], [0, 0], 'k--', alpha=0.3, label='轨道')

        ax.set_xlabel('轨道位置 (m)')

        ax.set_ylabel('高度 (m)')

        ax.set_title('工艺管廊点位分布')

        ax.legend(['轨道', '温度', '压力', '流量'], loc='upper right')

        ax.grid(True, alpha=0.3)

 

        # 2. 机器人轨迹

        ax = axes[0, 1]

        traj_pos = [t[0] for t in robot_trajectory]

        traj_vel = [t[1] for t in robot_trajectory]

 

        ax.plot(traj_pos, traj_vel, 'b-', linewidth=2, label='速度')

        ax.axhline(y=0, color='k', linestyle='--', alpha=0.3)

        for wp in route:

            ax.axvline(x=wp.position, color='r', linestyle=':', alpha=0.5)

        ax.set_xlabel('位置 (m)')

        ax.set_ylabel('速度 (m/s)')

        ax.set_title('机器人运动轨迹')

        ax.legend()

        ax.grid(True, alpha=0.3)

 

        # 3. 采集数据趋势

        ax = axes[1, 0]

        temps = [r['value'] for r in readings if r['unit'] == '°C']

        press = [r['value'] for r in readings if r['unit'] == 'MPa']

        flows = [r['value'] for r in readings if r['unit'] == 'm³/h']

 

        if temps:

            ax.plot(temps, 'r-o', label='温度(°C)', alpha=0.7)

        if press:

            ax.plot(press, 'b-s', label='压力(MPa)', alpha=0.7)

        if flows:

            ax.plot(flows, 'g-^', label='流量(m³/h)', alpha=0.7)

 

        ax.set_xlabel('采样序号')

        ax.set_ylabel('数值')

        ax.set_title('仪表采集数据')

        ax.legend()

        ax.grid(True, alpha=0.3)

 

        # 4. 状态统计

        ax = axes[1, 1]

        status_counts = {}

        for r in readings:

            status = r['status']

            status_counts[status] = status_counts.get(status, 0) + 1

 

        if status_counts:

            labels = list(status_counts.keys())

            sizes = list(status_counts.values())

            colors = ['green' if 'NORMAL' in l else

                     'orange' if 'WARNING' in l else 'red' for l in labels]

            ax.pie(sizes, labels=labels, autopct='%1.1f%%', colors=colors, startangle=90)

            ax.set_title('数据状态分布')

 

        plt.suptitle('工艺管道智能巡检仿真报告', fontsize=16, fontweight='bold')

        plt.tight_layout()

        plt.savefig(output, dpi=150, bbox_inches='tight')

        plt.close()

 

4.9 巡检机器人本体（状态模式）

 

class InspectionRobot:

    """

    巡检机器人本体 —— 状态模式

 

    状态流转: IDLE → MOVING → DWELLING → COLLECTING → IDLE

    """

 

    class State(Enum):

        IDLE = auto()

        MOVING = auto()

        DWELLING = auto() # 减速到位

        COLLECTING = auto() # 采集数据

        ERROR = auto()

 

    def __init__(self, motion_model: RobotMotionModel,

                 sensor: SensorSimulator,

                 recorder: DataRecorder):

        self.motion = motion_model

        self.sensor = sensor

        self.recorder = recorder

        self.pid = PIDPositionController()

        self.state = self.State.IDLE

        self.current_waypoint: Optional[Waypoint] = None

        self.dwell_start_time: float = 0.0

        self.collected_data: List[Dict] = []

        self.trajectory: List[Tuple[float, float]] = [] # (位置, 速度)

 

    def navigate_to(self, waypoint: Waypoint):

        """导航到指定点位"""

        self.current_waypoint = waypoint

        self.motion.set_target(waypoint.position)

        self.state = self.State.MOVING

        self.dwell_start_time = 0.0

 

    def update(self, dt: float, current_time: float):

        """更新机器人状态"""

        # 记录轨迹

        status = self.motion.get_status()

        self.trajectory.append((status['position'], status['velocity']))

 

        if self.state == self.State.MOVING:

            # 使用PID精确定位

            vel_cmd = self.pid.update(

                self.motion.target_position,

                self.motion.position,

                dt

            )

            # 实际更新运动模型

            pos, vel = self.motion.update(dt)

 

            # 检查是否到达

            if abs(pos - self.motion.target_position) < 0.01 and abs(vel) < 0.01:

                self.state = self.State.DWELLING

                self.dwell_start_time = current_time

 

        elif self.state == self.State.DWELLING:

            # 停留稳定

            if current_time - self.dwell_start_time >= self.current_waypoint.dwell_time * 0.5:

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