电力巡检实战：5天搞定自动化系统的速查手册

别再对着教程发呆，看了一堆视频还是不会写项目，才是最大的坑。很多人以为电力巡检系统很玄乎，其实核心逻辑就是“数据采集+规则判断+告警推送”。这篇速查手册不讲虚的，直接带你从零搭建一个可运行的最小可行产品（MVP），把代码跑通，比看十篇理论强。

项目目标与架构拆解

很多中小施工企业负责人容易陷入误区，觉得电力巡检必须是高精尖的AI视觉识别。其实对于日常运维来说，90%的场景是基于状态数据的逻辑判断。比如电压是否超限、温度是否异常、设备是否在线。

我们的项目目标很明确：

模拟数据源：用Python生成模拟的传感器数据（电压、电流、温度）。
核心引擎：编写一个轻量级的巡检引擎，实时处理数据流。
告警机制：当指标异常时，通过HTTP请求推送通知。
数据持久化：将巡检结果存入SQLite，方便后续查询。

这里要特别强调一点，很多初学者喜欢直接上Kafka、K8s、微服务，结果环境配了一天，代码还没写一行。工程化不等于复杂化，在单体应用中把逻辑闭环跑通，才是正道。架构上我们采用典型的三层结构：

采集层：模拟硬件接口，定时获取数据。
处理层：核心业务逻辑，包括阈值判断和状态机。
服务层：对外提供API接口，接收告警回调。

这种结构清晰，便于后续扩展。如果未来需要接入真实设备，只需替换采集层的驱动即可，核心逻辑无需改动。

目录结构与环境初始化

工欲善其事，必先利其器。一个规范的目录结构能帮你避免后期的代码混乱。建议采用以下结构：

power-inspection/
├── main.py          # 程序入口
├── config.py        # 配置文件
├── requirements.txt # 依赖库
├── core/
│   ├── __init__.py
│   ├── engine.py    # 核心巡检引擎
│   └── rules.py     # 规则定义
├── data/
│   ├── __init__.py
│   ├── simulator.py # 数据模拟器
│   └── storage.py   # 数据库操作
└── utils/├── __init__.py└── logger.py    # 日志工具

先创建虚拟环境，这是Python开发的基本功，能避免全局依赖污染：

python -m venv venv
source venv/bin/activate  # Windows用户用 venv\Scripts\activate
pip install requests sqlalchemy flask

requirements.txt 内容如下，锁定版本避免环境不一致：

requests==2.31.0
sqlalchemy==2.0.23
flask==3.0.0

注意：这里没有引入复杂的异步框架，因为对于中小规模的巡检任务，同步阻塞模型完全够用，且调试更简单。如果你的并发量达到万级，再考虑引入Celery或Asyncio，现在引入只会增加认知负担。

核心代码实现详解

1. 配置与规则定义

在 config.py 中定义全局配置，不要硬编码。

# config.py
class Config:# 巡检阈值VOLTAGE_LIMIT = 250  # 电压上限TEMP_LIMIT = 60      # 温度上限# 数据库连接DB_URL = "sqlite:///inspection.db"# 告警回调地址ALERT_URL = "http://localhost:5000/alert"

在 core/rules.py 中定义判断规则。这里采用策略模式的思想，将不同设备的规则解耦。

# core/rules.py
class Rule:def check(self, data: dict) -> bool:"""检查数据是否触发告警:param data: 包含 device_id, voltage, temp 的字典:return: True表示正常，False表示异常"""# 默认通过，具体子类重写return Trueclass SubstationRule(Rule):def check(self, data: dict) -> bool:from config import Config# 电压超限或温度超限，返回Falseif data['voltage'] > Config.VOLTAGE_LIMIT:return Falseif data['temp'] > Config.TEMP_LIMIT:return Falsereturn True

2. 数据模拟器

真实项目中，数据来自Modbus或MQTT。这里我们用随机数模拟，方便快速验证逻辑。

# data/simulator.py
import random
import timeclass Simulator:def __init__(self, device_id: str):self.device_id = device_idself.base_voltage = 220self.base_temp = 40def get_data(self) -> dict:# 模拟波动voltage = self.base_voltage + random.uniform(-10, 30)temp = self.base_temp + random.uniform(-5, 25)return {"device_id": self.device_id,"voltage": round(voltage, 2),"temp": round(temp, 2),"timestamp": time.time()}

3. 核心巡检引擎

这是项目的灵魂，core/engine.py。它负责调度模拟器、执行规则、存储数据并发送告警。

# core/engine.py
import time
import requests
from data.simulator import Simulator
from core.rules import SubstationRule
from data.storage import DBManager
from utils.logger import get_loggerlogger = get_logger("Engine")class InspectionEngine:def __init__(self):self.db = DBManager()self.rule = SubstationRule()self.simulator = Simulator("SUB-001")def run_once(self):"""执行一次巡检循环"""# 1. 获取数据data = self.simulator.get_data()logger.info(f"收到数据: {data}")# 2. 执行规则判断is_normal = self.rule.check(data)# 3. 存储结果self.db.save_record(data, is_normal)# 4. 如果异常，发送告警if not is_normal:self._send_alert(data)def _send_alert(self, data: dict):"""发送HTTP告警"""from config import Configtry:# 这里模拟发送POST请求# 实际项目中可能需要签名、重试机制logger.warning(f"触发告警: {data['device_id']}")# requests.post(Config.ALERT_URL, json=data)passexcept Exception as e:logger.error(f"告警发送失败: {e}")def start(self):"""启动巡检循环"""logger.info("巡检引擎启动...")while True:try:self.run_once()except Exception as e:logger.error(f"巡检循环异常: {e}")time.sleep(5)  # 每5秒巡检一次

4. 数据库与日志

data/storage.py 使用SQLAlchemy简化ORM操作。

# data/storage.py
from sqlalchemy import create_engine, Column, String, Float, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from config import Config
from datetime import datetimeBase = declarative_base()class InspectionRecord(Base):__tablename__ = 'records'id = Column(String, primary_key=True)device_id = Column(String)voltage = Column(Float)temp = Column(Float)is_normal = Column(Boolean)created_at = Column(DateTime, default=datetime.utcnow)class DBManager:def __init__(self):self.engine = create_engine(Config.DB_URL)Base.metadata.create_all(self.engine)self.Session = sessionmaker(bind=self.engine)def save_record(self, data: dict, is_normal: bool):session = self.Session()try:record = InspectionRecord(id=f"{data['device_id']}-{int(data['timestamp'])}",device_id=data['device_id'],voltage=data['voltage'],temp=data['temp'],is_normal=is_normal)session.add(record)session.commit()except Exception as e:session.rollback()raise efinally:session.close()

运行与测试验证

代码写完不能只看，必须跑。在 main.py 中启动服务。

# main.py
from core.engine import InspectionEngineif __name__ == "__main__":engine = InspectionEngine()engine.start()

运行 python main.py，观察控制台日志。你会发现，大部分时间状态是正常的，偶尔会出现“触发告警”的日志。

测试要点：

检查数据库：打开 inspection.db，查看 records 表，确认数据是否正确写入，is_normal 字段是否随数据波动而变化。
模拟故障：修改 simulator.py 中的 random.uniform 范围，让电压持续高于250，观察告警频率是否符合预期。
网络异常处理：在 _send_alert 中故意制造一个错误的URL，观察日志是否捕获了异常且程序没有崩溃。

很多新手在这里容易忽略异常处理。在工业现场，网络抖动是常态，如果因为一次请求失败导致整个巡检进程挂掉，那是严重的事故。务必在关键路径加上 try-except 块，并记录详细日志。

进阶技巧与避坑指南

1. 关于并发与性能

有人问，如果我有1000个设备，5秒巡检一次，会不会卡死？答案是：取决于你的单线程处理能力。Python的GIL限制使得CPU密集型任务无法真正并行。但对于IO密集型（如网络请求、数据库写入），可以使用 threading 或 asyncio。建议：初期先用单线程跑通逻辑，监控CPU和内存占用。如果瓶颈在IO，再引入 concurrent.futures.ThreadPoolExecutor 并行处理多个设备。不要过度设计。