巡检机器人阀门状态识别与自动读数仿真 —— 基于图像特征与状态机的OOP实战

 

“那年大夜班，零下二十度，巡检工拿着防爆手电在管廊里挨个核对阀门。回来后冻得直哆嗦，还说有个阀位看不清，不敢确定是全开还是微开。后来我们在防爆机器人上加了视觉模块，算法能精准识别手轮角度、阀杆位移，甚至能通过指针式压力表读数。现在中控室里喝着茶，就能知道现场几百个阀门的状态。”

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

 

一、实际应用场景描述

 

在石化、化工、核电等流程工业现场，阀门是控制介质流动的核心执行机构。传统的定期人工巡检存在诸多弊端：

 

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

│ 智能巡检机器人阀门识别系统 │

│ │

│ [防爆巡检机器人底盘] │

│ │ 自主导航 / SLAM │

│ ▼ │

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

│ │ 云台与视觉模块 │ │

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

│ │ │ 1. 相机标定 │ │ │

│ │ │ (畸变校正/像素比) │ │ │

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

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

│ │ │ 2. 目标检测 │ │ │

│ │ │ (YOLO/HOG/模板) │ │ │

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

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

│ │ │ 3. 特征提取 │ │ │

│ │ │ (轮廓/角度/刻度) │ │ │

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

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

│ │ 图像/特征向量 │

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

│ ▼ ▼ │

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

│ │ 状态判别器 │ │ 读数识别器 │ │

│ │ (Classifier)│ │ (OCR/Digit)│ │

│ │ • 手轮角度 │ │ • 表盘读数 │ │

│ │ • 阀杆位移 │ │ • 刻度识别 │ │

│ │ • 开关状态 │ │ • 单位换算 │ │

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

│ │ 状态标签 │ 数值结果 │

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

│ ▼ │

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

│ │ 上位机监控系统 │ │

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

│ │ │ 4. 数据融合 │ │ │

│ │ │ (视觉+DCS反馈) │ │ │

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

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

│ │ │ 5. 异常告警 │ │ │

│ │ │ (阀位不一致/卡涩) │ │ │

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

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

│ │ │ 6. 趋势记录 │ │ │

│ │ │ (开关时间/频次) │ │ │

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

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

│ │ Modbus/TCP, MQTT │

│ ▼ │

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

│ │ 中控室 SCADA/DCS │ │

│ │ • 实时阀位显示 │ │

│ │ • 历史趋势曲线 │ │

│ │ • 设备健康度评估 │ │

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

│ │

│ 核心: 图像处理 + 几何特征 + 状态机 + 冗余校验 │

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

 

人工巡检 vs 机器人巡检

 

维度 人工巡检 机器人视觉巡检

安全性 ❌ 高风险环境 ✅ 本质安全

一致性 ❌ 主观误差大 ✅ 客观量化

时效性 ❌ 几小时一次 ✅ 分钟级/实时

数据记录 ❌ 纸质/手抄 ✅ 数字化/可追溯

恶劣环境 ❌ 受限 ✅ 全天候

 

二、引入痛点

 

2.1 现场的真实困境

 

场景 现场发生了什么 根因

“看花了眼” “以为是开的，结果是关的” 视角盲区/光线差

“冻僵的手” “零下20度抄表半小时” 环境恶劣

“漏检隐患” “阀门微漏没发现” 巡检频次不足

“数据造假” “没去现场就填了表” 管理漏洞

“DCS不一致” “现场开着，画面显示关” 反馈信号故障

 

2.2 核心矛盾

 

阀门状态是“位置量”而非简单的“通断量”。 很多阀门（如球阀、蝶阀）在全开/全关之间有过渡区；而闸阀则靠阀杆升降。单纯依靠限位开关容易误判，且无法识别“卡在中间”或“虚假反馈”的情况。解决方案是引入机器视觉，通过识别手轮旋转角度或阀杆伸出长度，结合指针仪表读数，实现双重校验。

2.3 我们要解决什么

 

用一段精简的 Python 程序，构建一个巡检机器人阀门状态识别仿真系统，实现：

 

1. 阀门模型 —— 模拟不同种类阀门（手轮角行程、阀杆直行程）

2. 视觉模拟 —— 模拟相机拍摄、加入噪声与畸变

3. 特征提取 —— 基于几何特征（角度、位移）判断状态

4. 状态机判别 —— 结合阈值判定全开/全关/中间/故障

5. 读数识别 —— 模拟指针仪表读数（模拟OCR）

6. 可视化 —— 展示识别过程与结果

 

三、核心逻辑讲解

 

3.1 阀门状态识别原理

 

本工具基于哈工程《工业过程控制》第九章“执行器”和计算机视觉基础：

 

① 角行程阀门（手轮/球阀/蝶阀）

 

通过检测手轮辐条或指示箭头的角度：

 

Angle = \arctan2(p_y - c_y, p_x - c_x) \times \frac{180}{\pi}

 

状态判定：

 

- 全关 (Closed): |Angle - Angle_{closed}| < \epsilon 

- 全开 (Open): |Angle - Angle_{open}| < \epsilon 

- 中间位 (Mid): 介于两者之间

 

② 直行程阀门（闸阀/截止阀）

 

通过检测阀杆端部的垂直位移：

 

Position = y_{top} - y_{bottom}

 

状态判定：

 

- 全关: Position < Threshold_{min} 

- 全开: Position > Threshold_{max} 

 

③ 指针仪表读数

 

基于刻度盘圆心和指针端点计算角度，映射到量程：

 

Value = \frac{Angle - Angle_{zero}}{Angle_{full} - Angle_{zero}} \times (Range_{max} - Range_{min}) + Range_{min}

 

3.2 状态机设计

 

        ┌─────────┐

        │ UNKNOWN │

        └────┬────┘

             │ 首次检测

             ▼

        ┌─────────┐◀────┐

        │ CLOSED │ │ 置信度<阈值

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

             │ 角度变化 │

             ▼ │

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

        │ MIDWAY │─────┘

        └────┬────┘

             │ 角度到位

             ▼

        ┌─────────┐

        │ OPEN │

        └────┬────┘

             │ 与DCS反馈不一致

             ▼

        ┌─────────┐

        │ FAULT │

        └─────────┘

 

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

 

4.1 类结构总览

 

类名 职责 设计模式

 

"ValveType" 阀门类型枚举 枚举

 

"ValveState" 阀门状态枚举 枚举

 

"Point2D" 二维点（值对象） 值对象

 

"ImageFrame" 模拟图像帧 封装

 

"ValveConfig" 阀门配置（值对象） 值对象

 

"ValveModel" 阀门物理模型 策略模式

 

"CameraSensor" 相机传感器模型 封装

 

"FeatureExtractor" 特征提取器 模板方法

 

"StateClassifier" 状态判别器 策略模式

 

"GaugeReader" 仪表读数器 策略模式

 

"InspectionRobot" 巡检机器人（聚合根） 聚合根

 

"VisualizationEngine" 可视化引擎 封装

 

4.2 数据模型与枚举

 

from dataclasses import dataclass, field

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

from enum import Enum, auto

from collections import deque

import numpy as np

import matplotlib.pyplot as plt

from datetime import datetime

import math

import cv2 # 模拟导入，实际运行时若无cv2可注释相关部分

 

class ValveType(Enum):

    """阀门类型"""

    ANGLE_STROKE = auto() # 角行程（手轮/球阀/蝶阀）

    LINEAR_STROKE = auto() # 直行程（闸阀/截止阀）

    QUARTER_TURN = auto() # 1/4回转（球阀/蝶阀特例）

 

class ValveState(Enum):

    """阀门状态"""

    UNKNOWN = auto()

    FULLY_CLOSED = auto()

    FULLY_OPEN = auto()

    MID_POSITION = auto()

    STUCK = auto() # 卡涩

    FAULT = auto() # 故障（与DCS不一致）

 

@dataclass(frozen=True)

class Point2D:

    """二维点 —— 值对象"""

    x: float

    y: float

 

    def distance_to(self, other: 'Point2D') -> float:

        return math.hypot(self.x - other.x, self.y - other.y)

 

    def to_tuple(self) -> Tuple[int, int]:

        return (int(self.x), int(self.y))

 

@dataclass

class ValveConfig:

    """阀门配置 —— 值对象"""

    valve_id: str

    valve_type: ValveType

    center_point: Point2D # 旋转中心或底部参考点

    closed_angle: float = 0.0 # 全关位置角度（度）

    open_angle: float = 90.0 # 全开位置角度（度）

    full_stroke: float = 100.0 # 直行程全长（mm）

    tolerance: float = 5.0 # 判定容差（度或mm）

    gauge_center: Optional[Point2D] = None # 附属仪表中心

    gauge_zero_angle: float = -135.0 # 表盘0度位置

    gauge_full_angle: float = 135.0 # 表盘满度位置

    gauge_range: Tuple[float, float] = (0.0, 1.0) # 量程

 

4.3 阀门物理模型（策略模式）

 

class ValveModel:

    """

    阀门物理模型 —— 策略模式

 

    模拟阀门的物理运动和当前状态

    """

 

    def __init__(self, config: ValveConfig):

        self.config = config

        self.current_angle: float = config.closed_angle # 当前角度

        self.current_position: float = 0.0 # 当前位移（直行程）

        self.target_angle: float = config.closed_angle

        self.target_position: float = 0.0

        self.state: ValveState = ValveState.FULLY_CLOSED

        self.history: List[Tuple[float, ValveState, float]] = []

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

 

    def set_target(self, target_state: ValveState):

        """设置目标状态"""

        if target_state == ValveState.FULLY_OPEN:

            self.target_angle = self.config.open_angle

            self.target_position = self.config.full_stroke

        elif target_state == ValveState.FULLY_CLOSED:

            self.target_angle = self.config.closed_angle

            self.target_position = 0.0

        # MID_POSITION 需要外部指定具体角度/位置

 

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

        """

        更新阀门状态（模拟物理运动）

 

        Args:

            dt: 时间步长

 

        Returns:

            当前状态

        """

        speed = 30.0 # 度/秒 或 mm/秒

 

        if self.config.valve_type in [ValveType.ANGLE_STROKE, ValveType.QUARTER_TURN]:

            # 角度运动

            delta = (self.target_angle - self.current_angle)

            if abs(delta) > 0.1:

                move = math.copysign(min(speed * dt, abs(delta)), delta)

                self.current_angle += move

            else:

                self.current_angle = self.target_angle

 

            # 状态判定

            if abs(self.current_angle - self.config.closed_angle) < self.config.tolerance:

                self.state = ValveState.FULLY_CLOSED

            elif abs(self.current_angle - self.config.open_angle) < self.config.tolerance:

                self.state = ValveState.FULLY_OPEN

            else:

                self.state = ValveState.MID_POSITION

 

        elif self.config.valve_type == ValveType.LINEAR_STROKE:

            # 直线位移

            delta = (self.target_position - self.current_position)

            if abs(delta) > 0.1:

                move = math.copysign(min(speed * dt, abs(delta)), delta)

                self.current_position += move

            else:

                self.current_position = self.target_position

 

            # 状态判定

            if self.current_position < self.config.tolerance:

                self.state = ValveState.FULLY_CLOSED

            elif abs(self.current_position - self.config.full_stroke) < self.config.tolerance:

                self.state = ValveState.FULLY_OPEN

            else:

                self.state = ValveState.MID_POSITION

 

        self.history.append((datetime.now().timestamp(), self.state, self.current_angle))

        return self.state

 

    def get_current_feature(self) -> Dict[str, Any]:

        """获取当前特征（用于视觉识别）"""

        if self.config.valve_type in [ValveType.ANGLE_STROKE, ValveType.QUARTER_TURN]:

            return {

                "type": "angle",

                "value": self.current_angle,

                "center": self.config.center_point

            }

        else:

            return {

                "type": "position",

                "value": self.current_position,

                "base": self.config.center_point

            }

 

    def inject_fault(self, fault_type: str = "stuck"):

        """注入故障（用于测试）"""

        if fault_type == "stuck":

            self.target_angle = self.current_angle # 停止运动

            self.target_position = self.current_position

            self.state = ValveState.STUCK

 

4.4 相机传感器模型

 

class CameraSensor:

    """

    相机传感器模型

 

    模拟图像采集、噪声、畸变

    """

 

    def __init__(self, resolution: Tuple[int, int] = (640, 480), noise_level: float = 0.02):

        self.width, self.height = resolution

        self.noise_level = noise_level

        self.focal_length = 800.0

        self.principal_point = Point2D(resolution[0]/2, resolution[1]/2)

 

    def capture(self, valve_model: ValveModel) -> 'ImageFrame':

        """

        模拟拍摄阀门图像

 

        Args:

            valve_model: 阀门模型

 

        Returns:

            模拟图像帧

        """

        # 创建空白图像

        img = np.ones((self.height, self.width, 3), dtype=np.uint8) * 240

 

        feature = valve_model.get_current_feature()

 

        if feature["type"] == "angle":

            # 绘制手轮/指针

            center = feature["center"]

            angle_rad = math.radians(feature["value"])

 

            # 加入噪声

            noisy_angle = angle_rad + np.random.normal(0, self.noise_level * 0.1)

            noisy_center = Point2D(

                center.x + np.random.normal(0, self.noise_level * 5),

                center.y + np.random.normal(0, self.noise_level * 5)

            )

 

            # 计算指针端点

            length = 80

            end_x = noisy_center.x + length * math.cos(noisy_angle)

            end_y = noisy_center.y + length * math.sin(noisy_angle)

 

            # 绘制

            cv2.circle(img, noisy_center.to_tuple(), 30, (200, 200, 200), -1) # 轮毂

            cv2.line(img, noisy_center.to_tuple(), (int(end_x), int(end_y)),

                     (0, 0, 255), 3) # 指针

            cv2.circle(img, noisy_center.to_tuple(), 5, (0, 0, 0), -1) # 中心轴

 

        elif feature["type"] == "position":

            # 绘制阀杆

            base = feature["base"]

            height = feature["value"] * 0.8 # 缩放因子

 

            # 加入噪声

            noisy_base = Point2D(

                base.x + np.random.normal(0, self.noise_level * 3),

                base.y + np.random.normal(0, self.noise_level * 3)

            )

            noisy_height = height + np.random.normal(0, self.noise_level * 2)

 

            top_y = noisy_base.y - noisy_height

 

            # 绘制阀体

            cv2.rectangle(img,

                         (int(noisy_base.x - 15), int(noisy_base.y)),

                         (int(noisy_base.x + 15), int(noisy_base.y + 50)),

                         (100, 100, 100), -1)

            # 绘制阀杆

            cv2.rectangle(img,

                         (int(noisy_base.x - 5), int(top_y)),

                         (int(noisy_base.x + 5), int(noisy_base.y)),

                         (0, 255, 0), -1)

 

        # 添加高斯噪声

        noise = np.random.normal(0, self.noise_level * 255, img.shape).astype(np.int16)

        img = np.clip(img.astype(np.int16) + noise, 0, 255).astype(np.uint8)

 

        return ImageFrame(img, datetime.now().timestamp(), valve_model.config.valve_id)

 

4.5 特征提取与状态判别

 

class FeatureExtractor:

    """

    特征提取器 —— 模板方法模式

 

    从图像中提取阀门特征

    """

 

    def extract_features(self, frame: 'ImageFrame', valve_config: ValveConfig) -> Dict[str, Any]:

        """

        提取特征（模拟版本，实际应使用CV算法）

 

        Args:

            frame: 图像帧

            valve_config: 阀门配置

 

        Returns:

            提取到的特征字典

        """

        # 在真实系统中，这里会是复杂的CV算法：

        # 1. 预处理 (灰度化、滤波、边缘检测)

        # 2. 轮廓查找 (findContours)

        # 3. 几何拟合 (fitEllipse, minAreaRect)

        # 4. 关键点检测 (HoughLines, HoughCircles)

 

        # 由于这是仿真，我们直接从"完美"的模拟图像中反推特征

        # 实际应用中，这里是最复杂的部分

 

        img = frame.image

        h, w = img.shape[:2]

 

        if valve_config.valve_type in [ValveType.ANGLE_STROKE, ValveType.QUARTER_TURN]:

            # 寻找红色指针（简单颜色阈值）

            hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)

            lower_red = np.array([0, 100, 100])

            upper_red = np.array([10, 255, 255])

            mask = cv2.inRange(hsv, lower_red, upper_red)

 

            # 查找轮廓

            contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)

 

            if contours:

                # 找到最长的轮廓（指针）

                cnt = max(contours, key=cv2.contourArea)

                # 拟合直线

                [vx, vy, x, y] = cv2.fitLine(cnt, cv2.DIST_L2, 0, 0.01, 0.01)

                # 计算角度

                angle = math.atan2(vy, vx) * 180 / math.pi

                # 调整到0-360度

                if angle < 0:

                    angle += 360

 

                return {

                    "type": "angle",

                    "raw_angle": angle,

                    "confidence": 0.9 # 模拟置信度

                }

 

        elif valve_config.valve_type == ValveType.LINEAR_STROKE:

            # 寻找绿色阀杆（简单颜色阈值）

            hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)

            lower_green = np.array([40, 50, 50])

            upper_green = np.array([80, 255, 255])

            mask = cv2.inRange(hsv, lower_green, upper_green)

 

            # 查找轮廓

            contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)

 

            if contours:

                cnt = max(contours, key=cv2.contourArea)

                x, y, w, h = cv2.boundingRect(cnt)

                position = h # 高度代表位移

 

                return {

                    "type": "position",

                    "raw_position": position,

                    "confidence": 0.85

                }

 

        return {"type": "unknown", "confidence": 0.0}

 

 

class StateClassifier:

    """

    状态判别器 —— 策略模式

 

    根据提取的特征判断阀门状态

    """

 

    def classify(self, features: Dict[str, Any], config: ValveConfig) -> Tuple[ValveState, float]:

        """

        分类阀门状态

 

        Args:

            features: 提取的特征

            config: 阀门配置

 

        Returns:

            (状态, 置信度)

        """

        if features["type"] == "angle":

            angle = features["raw_angle"]

            conf = features["confidence"]

 

            # 处理角度环绕（例如355度接近0度）

            def angle_diff(a1, a2):

                diff = abs(a1 - a2)

                return min(diff, 360 - diff)

 

            if angle_diff(angle, config.closed_angle) < config.tolerance:

                return ValveState.FULLY_CLOSED, conf

            elif angle_diff(angle, config.open_angle) < config.tolerance:

                return ValveState.FULLY_OPEN, conf

            else:

                return ValveState.MID_POSITION, conf * 0.8 # 中间位置置信度稍低

 

        elif features["type"] == "position":

            pos = features["raw_position"]

            conf = features["confidence"]

 

            # 归一化位置（假设图像中100像素代表全行程）

            normalized_pos = (pos / 100.0) * config.full_stroke

 

            if normalized_pos < config.tolerance:

                return ValveState.FULLY_CLOSED, conf

            elif abs(normalized_pos - config.full_stroke) < config.tolerance:

                return ValveState.FULLY_OPEN, conf

            else:

                return ValveState.MID_POSITION, conf * 0.8

 

        return ValveState.UNKNOWN, 0.0

 

 

class GaugeReader:

    """

    指针仪表读数器 —— 策略模式

 

    读取附属的压力/流量/液位表

    """

 

    def read_gauge(self, frame: 'ImageFrame', config: ValveConfig) -> Optional[float]:

        """

        读取仪表数值

 

        Args:

            frame: 图像帧

            config: 阀门配置（包含仪表信息）

 

        Returns:

            读数值，失败返回None

        """

        if not config.gauge_center:

            return None

 

        # 模拟读数：基于阀门状态估算

        # 实际应使用与FeatureExtractor类似的指针检测算法

        if config.valve_type == ValveType.ANGLE_STROKE:

            # 假设球阀开度与流量成正比

            angle_ratio = (config.open_angle - config.closed_angle)

            if angle_ratio != 0:

                current_ratio = (frame.valve_model.current_angle - config.closed_angle) / angle_ratio

                value = config.gauge_range[0] + current_ratio * (config.gauge_range[1] - config.gauge_range[0])

                # 加入测量噪声

                value += np.random.normal(0, 0.01 * (config.gauge_range[1] - config.gauge_range[0]))

                return max(config.gauge_range[0], min(value, config.gauge_range[1]))

 

        return None

 

4.6 图像帧与巡检机器人（聚合根）

 

@dataclass

class ImageFrame:

    """图像帧 —— 值对象"""

    image: np.ndarray

    timestamp: float

    valve_id: str

    metadata: Dict = field(default_factory=dict)

 

 

class InspectionRobot:

    """

    巡检机器人 —— 聚合根

 

    协调整个巡检流程

    """

 

    def __init__(self, robot_id: str = "ROBOT-01"):

        self.robot_id = robot_id

        self.camera = CameraSensor()

        self.extractor = FeatureExtractor()

        self.classifier = StateClassifier()

        self.gauge_reader = GaugeReader()

 

        self.valves: Dict[str, ValveModel] = {}

        self.dcs_states: Dict[str, ValveState] = {} # 模拟DCS反馈

 

        self.inspection_history: List[Dict] = []

        self.alarm_log: List[Dict] = []

 

        # 统计信息

        self.stats = {

            "total_inspections": 0,

            "successful_reads": 0,

            "fault_detections": 0,

            "discrepancies": 0

        }

 

    def add_valve(self, config: ValveConfig, dcs_state: ValveState = ValveState.FULLY_CLOSED):

        """添加阀门到巡检列表"""

        model = ValveModel(config)

        model.state = dcs_state

        self.valves[config.valve_id] = model

        self.dcs_states[config.valve_id] = dcs_state

        print(f"🔧 注册阀门: {config.valve_id} ({config.valve_type.name})")

 

    def inspect_valve(self, valve_id: str) -> Dict:

        """

        巡检单个阀门

 

        Args:

            valve_id: 阀门ID

 

        Returns:

            巡检结果字典

        """

        if valve_id not in self.valves:

            raise ValueError(f"阀门 {valve_id} 未注册")

 

        valve = self.valves[valve_id]

        config = valve.config

 

        # 1. 模拟物理更新（如果阀门正在动作）

        valve.update(0.1)

 

        # 2. 拍摄图像

        frame = self.camera.capture(valve)

 

        # 3. 提取特征

        features = self.extractor.extract_features(frame, config)

 

        # 4. 分类状态

        vision_state, confidence

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