电力中心配电柜监测：SNMP协议读取以太网温湿度记录仪，告警规则配置

原创

HONSOR盛世宏博

发布于 2026-09-24 15:36:52

570

电力中心配电柜监测：SNMP协议读取以太网温湿度记录仪，告警规则配置

关键词：电力中心、配电柜、SNMP协议、以太网温湿度记录仪、MIB库、Trap告警、阈值配置、OID、轮询策略、配电室环境监测标签：#物联网 #TCP/IP #UDP #POE供电 #以太网温湿度传感器 #网口温湿度变送器 #机房监控 #SNMP #配电柜监测

一、电力中心配电柜场景的特殊性

电力中心（变电站、配电室、开闭所、新能源升压站）的配电柜内环境监控，与机房动环、楼宇自控、产线监测存在显著差异：

维度	机房动环	产线监测	电力中心配电柜
电磁环境	中等	强干扰	极强：高压开关、变压器、电流互感器、电弧
空间约束	机柜内	车间区域	极窄：配电柜内/柜顶，空间受限
安装方式	导轨/壁挂	壁挂/支架	磁吸/卡扣/超薄，不占柜内空间
通信要求	稳定即可	防风暴	抗电磁干扰，光纤优先
告警要求	分级通知	快速恢复	联动跳闸/风机/消防，安全等级最高
运维方式	定期巡检	产线不停	带电作业，不能断电插拔
协议偏好	Modbus TCP/HTTP	TCP/UDP	SNMP（电力监控网络标准协议）

电力行业监控系统（SCADA/动环监控主机）普遍支持SNMP协议，用于统一采集网络设备、UPS、空调、环境传感器的数据。以太网温湿度记录仪支持SNMP后，可直接接入现有电力监控网络，无需额外网关或协议转换。

二、SNMP协议基础与记录仪MIB解析

2.1 SNMP版本选择

版本	安全性	复杂度	电力场景适用性
SNMPv1	无认证，明文	最低	不推荐
SNMPv2c	社区字符串（明文）	低	内网隔离场景可用
SNMPv3	用户认证+加密（AES/DES）	中	电力监控首选

配电柜环境数据涉及电力设施运行状态，建议启用SNMPv3的authPriv模式（认证+加密），防止数据被窃听或篡改。

2.2 记录仪MIB库结构

以太网温湿度记录仪通常提供私有MIB文件（.mib），定义设备的OID树。典型结构：

.iso.org.dod.internet.private.enterprises.{enterpriseID}.{productID}
├── system          (1)  系统信息：设备名、位置、描述
├── sensor          (2)  传感器数据
│   ├── temperature (1)  温度
│   │   ├── current (1)  当前值（0.1℃精度）
│   │   ├── min     (2)  最小值
│   │   ├── max     (3)  最大值
│   │   └── avg     (4)  平均值
│   ├── humidity    (2)  湿度
│   │   ├── current (1)
│   │   ├── min     (2)
│   │   ├── max     (3)
│   │   └── avg     (4)
│   └── dewpoint    (3)  露点
├── config          (3)  配置参数
│   ├── tempHigh    (1)  温度上限阈值
│   ├── tempLow     (2)  温度下限阈值
│   ├── humHigh     (3)  湿度上限阈值
│   ├── humLow      (4)  湿度下限阈值
│   └── pollInterval(5)  采集周期
├── status          (4)  设备状态
│   ├── alarmStatus (1)  告警状态位
│   ├── powerStatus (2)  供电状态
│   └── networkStatus(3) 网络状态
└── trap            (5)  Trap配置
    ├── trapEnable  (1)
    ├── trapServer  (2)  Trap目标IP
    └── trapPort    (3)  Trap目标端口

2.3 加载MIB并读取数据

# 使用snmpwalk读取设备所有可用数据
snmpwalk -v2c -c public 192.168.200.101 .1.3.6.1.4.1.12345.1

# 读取当前温度
snmpget -v2c -c public 192.168.200.101 .1.3.6.1.4.1.12345.1.2.1.1.0

# 使用SNMPv3（推荐）
snmpget -v3 -u monitor -l authPriv -a SHA -A authpass -x AES -X privpass \
  192.168.200.101 .1.3.6.1.4.1.12345.1.2.1.1.0

Python批量读取示例：

"""
snmp_poller.py - SNMP批量轮询配电柜温湿度记录仪
"""

from pysnmp.hlapi import *
from pysnmp.entity import config
import asyncio
from dataclasses import dataclass

@dataclass
class SNMPDevice:
    ip: str
    version: str = "v3"
    user: str = "monitor"
    auth_key: str = "authpass"
    priv_key: str = "privpass"
    community: str = "public"
    oid_tree: str = ".1.3.6.1.4.1.12345.1"

class SNMPTempHumidPoller:
    def __init__(self, devices: list[SNMPDevice], poll_interval=30):
        self.devices = devices
        self.poll_interval = poll_interval
        self.cache = {}
    
    def _build_auth(self, dev: SNMPDevice):
        """构建SNMP认证参数"""
        if dev.version == "v3":
            return UsmUserData(
                dev.user,
                dev.auth_key,
                dev.priv_key,
                authProtocol=config.usmHMACSHAAuthProtocol,
                privProtocol=config.usmAesCfb128Protocol
            )
        else:
            return CommunityData(dev.community, mpModel=1 if dev.version == "v2c" else 0)
    
    async def poll_device(self, dev: SNMPDevice):
        """轮询单台设备"""
        auth = self._build_auth(dev)
        transport = UdpTransportTarget((dev.ip, 161), timeout=2, retries=2)
        
        oids = {
            "temperature": ObjectIdentity(dev.oid_tree + ".2.1.1.0"),
            "humidity": ObjectIdentity(dev.oid_tree + ".2.2.1.0"),
            "dewpoint": ObjectIdentity(dev.oid_tree + ".2.3.1.0"),
            "alarm_status": ObjectIdentity(dev.oid_tree + ".4.1.1.0"),
        }
        
        results = {}
        for name, oid in oids.items():
            iterator = getCmd(
                SnmpEngine(),
                auth,
                transport,
                ContextData(),
                oid
            )
            errorIndication, errorStatus, errorIndex, varBinds = next(iterator)
            if errorIndication:
                print(f"{dev.ip} {name}: {errorIndication}")
                results[name] = None
            elif errorStatus:
                print(f"{dev.ip} {name}: {errorStatus.prettyPrint()}")
                results[name] = None
            else:
                for varBind in varBinds:
                    value = varBind[1]
                    # 温度值通常×10存储
                    if name == "temperature":
                        results[name] = float(value) / 10.0
                    elif name == "humidity":
                        results[name] = float(value) / 10.0
                    elif name == "dewpoint":
                        results[name] = float(value) / 10.0
                    else:
                        results[name] = int(value)
        
        results["_online"] = all(v is not None for v in results.values())
        self.cache[dev.ip] = results
        return results
    
    async def run(self):
        """主轮询循环"""
        while True:
            for dev in self.devices:
                asyncio.create_task(self.poll_device(dev))
            await asyncio.sleep(self.poll_interval)

三、告警规则配置

3.1 告警层级设计

配电柜环境告警应分层级，与电力监控系统告警级别对齐：

级别	名称	触发条件	响应要求	联动动作
Level 1	提示	温度>30℃或湿度>60%RH	记录日志，可忽略	无
Level 2	一般告警	温度>35℃或湿度>70%RH	24h内处理	启动柜顶风机
Level 3	严重告警	温度>40℃或湿度>80%RH	立即处理	启动风机+声光报警+通知运维
Level 4	紧急告警	温度>50℃或湿度>90%RH	立即停电检查	联动跳闸+消防联动+短信通知

3.2 记录仪本地告警配置

通过SNMP SET命令配置记录仪本地阈值：

"""
snmp_config.py - 通过SNMP配置记录仪告警阈值
"""

from pysnmp.hlapi import *

def set_threshold(dev_ip, oid_base, temp_high, temp_low, hum_high, hum_low):
    """设置温湿度告警阈值"""
    auth = UsmUserData("admin", "admin_auth", "admin_priv",
                       authProtocol=config.usmHMACSHAAuthProtocol,
                       privProtocol=config.usmAesCfb128Protocol)
    transport = UdpTransportTarget((dev_ip, 161), timeout=3, retries=3)
    
    # 阈值×10后写入（设备内部精度0.1）
    thresholds = [
        (oid_base + ".3.1.1.0", temp_high * 10),  # tempHigh
        (oid_base + ".3.1.2.0", temp_low * 10),   # tempLow
        (oid_base + ".3.1.3.0", hum_high * 10),   # humHigh
        (oid_base + ".3.1.4.0", hum_low * 10),    # humLow
    ]
    
    for oid, value in thresholds:
        setCmd(
            SnmpEngine(),
            auth,
            transport,
            ContextData(),
            ObjectType(ObjectIdentity(oid), Integer(value))
        )
        print(f"{dev_ip}: Set {oid} = {value}")

# 配置所有配电柜记录仪
devices = [
    {"ip": "192.168.200.101", "cabinet": "10kV进线柜A"},
    {"ip": "192.168.200.102", "cabinet": "10kV进线柜B"},
    {"ip": "192.168.200.103", "cabinet": "变压器柜A"},
    {"ip": "192.168.200.104", "cabinet": "变压器柜B"},
    {"ip": "192.168.200.105", "cabinet": "低压出线柜A"},
]

oid_base = ".1.3.6.1.4.1.12345.1"

for dev in devices:
    set_threshold(dev["ip"], oid_base, temp_high=40, temp_low=5, hum_high=80, hum_low=20)
    print(f"Configured {dev['cabinet']} ({dev['ip']})")

3.3 SNMP Trap配置

记录仪在阈值越限时主动发送SNMP Trap，无需轮询：

"""
snmp_trap_receiver.py - 接收并处理记录仪Trap
"""

from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher
from pysnmp.carrier.asyncore.dgram import udp
from pysnmp.proto import api
from pysnmp.entity import config as snmp_config
import datetime

class TrapReceiver:
    def __init__(self, listen_ip="0.0.0.0", listen_port=162):
        self.listen_ip = listen_ip
        self.listen_port = listen_port
        self.trap_handlers = []
    
    def add_handler(self, handler):
        """添加Trap处理函数"""
        self.trap_handlers.append(handler)
    
    def _handle_trap(self, transportDispatcher, transportDomain, data):
        """解析Trap"""
        msg = api.v1.decodeMessage(data)
        pdu = msg.getPDU()
        
        trap_info = {
            "timestamp": datetime.datetime.now().isoformat(),
            "source_ip": transportDomain[1][0],
            "trap_type": pdu.getComponentByName("generic-trap"),
            "specific_trap": pdu.getComponentByName("specific-trap"),
            "variables": {}
        }
        
        # 解析VarBinds
        for oid, value in pdu.getComponentByName("variable-bindings"):
            oid_str = str(oid.getComponent())
            trap_info["variables"][oid_str] = str(value.getComponent())
        
        # 调用所有处理函数
        for handler in self.trap_handlers:
            handler(trap_info)
    
    def start(self):
        """启动Trap接收器"""
        dispatcher = AsyncoreDispatcher()
        dispatcher.registerRecvCbFun(self._handle_trap)
        dispatcher.registerTransport(
            udp.domainName + (1,),
            udp.UdpTransport().openServerMode((self.listen_ip, self.listen_port))
        )
        print(f"Trap receiver listening on {self.listen_ip}:{self.listen_port}")
        dispatcher.runDispatcher()

# 告警处理函数
def alert_handler(trap_info):
    """处理告警Trap"""
    source_ip = trap_info["source_ip"]
    variables = trap_info["variables"]
    
    # 解析告警信息
    temp = variables.get(".1.3.6.1.4.1.12345.1.2.1.1.0", "N/A")
    hum = variables.get(".1.3.6.1.4.1.12345.1.2.2.1.0", "N/A")
    alarm_status = variables.get(".1.3.6.1.4.1.12345.1.4.1.1.0", "0")
    
    print(f"[{trap_info['timestamp']}] ALERT from {source_ip}")
    print(f"  Temperature: {temp}°C, Humidity: {hum}%RH")
    print(f"  Alarm Status: {alarm_status}")
    
    # 根据告警状态执行联动
    if int(alarm_status) & 0x01:  # 温度高告警
        print("  Action: Start exhaust fan, notify maintenance")
    if int(alarm_status) & 0x02:  # 温度低告警
        print("  Action: Start heater")
    if int(alarm_status) & 0x04:  # 湿度高告警
        print("  Action: Start dehumidifier")
    if int(alarm_status) & 0x08:  # 湿度低告警
        print("  Action: Start humidifier")

# 启动Trap接收
receiver = TrapReceiver(listen_port=162)
receiver.add_handler(alert_handler)
receiver.start()

3.4 联动规则引擎

"""
alarm_engine.py - 告警规则引擎
"""

from enum import IntEnum
from dataclasses import dataclass
from typing import Callable, List
import time

class AlarmLevel(IntEnum):
    INFO = 1
    WARNING = 2
    CRITICAL = 3
    EMERGENCY = 4

@dataclass
class AlarmRule:
    name: str
    condition: Callable[[dict], bool]
    level: AlarmLevel
    actions: List[Callable]
    debounce_seconds: float = 60.0  # 防抖时间
    auto_clear: bool = True

class AlarmEngine:
    def __init__(self):
        self.rules: List[AlarmRule] = []
        self.active_alarms = {}  # (rule_name, device_id) -> trigger_time
    
    def add_rule(self, rule: AlarmRule):
        self.rules.append(rule)
    
    def evaluate(self, device_id: str, readings: dict):
        """评估所有规则"""
        for rule in self.rules:
            key = (rule.name, device_id)
            triggered = rule.condition(readings)
            
            if triggered:
                if key not in self.active_alarms:
                    self.active_alarms[key] = time.time()
                    # 检查防抖
                    if time.time() - self.active_alarms[key] >= rule.debounce_seconds:
                        self._fire_alarm(rule, device_id, readings)
            else:
                if key in self.active_alarms and rule.auto_clear:
                    del self.active_alarms[key]
                    self._clear_alarm(rule, device_id)
    
    def _fire_alarm(self, rule: AlarmRule, device_id: str, readings: dict):
        """触发告警"""
        print(f"[ALARM-{rule.level.name}] {rule.name} on {device_id}")
        print(f"  Readings: {readings}")
        for action in rule.actions:
            action(device_id, readings, rule.level)
    
    def _clear_alarm(self, rule: AlarmRule, device_id: str):
        """清除告警"""
        print(f"[CLEARED] {rule.name} on {device_id}")

# 联动动作定义
def start_exhaust_fan(device_id, readings, level):
    print(f"  -> Starting exhaust fan for {device_id}")

def start_dehumidifier(device_id, readings, level):
    print(f"  -> Starting dehumidifier for {device_id}")

def send_sms_alert(device_id, readings, level):
    print(f"  -> Sending SMS alert for {device_id}")

def trigger_power_off(device_id, readings, level):
    print(f"  -> TRIGGERING POWER OFF for {device_id}")

# 配置告警规则
engine = AlarmEngine()

# 温度高告警
engine.add_rule(AlarmRule(
    name="HighTemperature",
    condition=lambda r: r.get("temperature", 0) > 40,
    level=AlarmLevel.CRITICAL,
    actions=[start_exhaust_fan, send_sms_alert],
    debounce_seconds=30
))

# 湿度高告警
engine.add_rule(AlarmRule(
    name="HighHumidity",
    condition=lambda r: r.get("humidity", 0) > 80,
    level=AlarmLevel.WARNING,
    actions=[start_dehumidifier],
    debounce_seconds=60
))

# 紧急温度告警
engine.add_rule(AlarmRule(
    name="EmergencyTemperature",
    condition=lambda r: r.get("temperature", 0) > 50,
    level=AlarmLevel.EMERGENCY,
    actions=[trigger_power_off, send_sms_alert],
    debounce_seconds=10
))

四、典型坑

SNMPv3用户配置错误：设备端和接收端用户/密码/加密算法不一致，导致Trap无法解密或GET失败。解决：先用SNMPv2c测试连通性，再切换到v3。
MIB文件未加载：网管系统无法识别私有OID，显示为数字串。解决：将MIB文件导入网管系统，确保OID解析正确。
Trap端口被防火墙阻断：UDP/162端口未开放，Trap丢失。解决：检查防火墙规则，确保162端口放行。
轮询间隔过短：大量设备频繁轮询，网络拥塞。解决：根据设备数量计算合理间隔，或使用Trap替代轮询。
阈值设置不合理：阈值过于敏感，频繁误报；或过于宽松，漏报。解决：基于历史数据统计分析，设置合理阈值。
告警风暴：多个设备同时触发告警，告警信息淹没。解决：实现告警聚合和抑制，相同告警合并通知。
设备离线误告警：网络中断导致设备离线，触发所有阈值告警。解决：区分设备离线告警和阈值告警，离线时不触发阈值告警。
联动动作未确认：联动跳闸后未确认，导致误操作。解决：关键联动动作需人工确认或二次验证。
记录仪本地存储满：长期运行后本地存储满，新数据无法写入。解决：配置循环覆盖或定期导出数据。
电磁干扰导致SNMP通信失败：配电柜内强电磁干扰导致UDP丢包。解决：使用屏蔽网线，远离强电线路，必要时光纤传输。

五、小结

电力中心配电柜监测中，SNMP协议读取以太网温湿度记录仪并配置告警规则，是电力监控系统集成的关键环节。通过SNMPv3保障数据安全，合理设计告警层级和联动规则，实现从数据采集到告警响应的完整闭环。配置本地阈值确保设备独立告警能力，同时接收SNMP Trap实现实时告警推送。告警规则引擎支持灵活配置和防抖处理，避免误报。配电柜环境恶劣，设备选型和安装需考虑电磁兼容性和空间约束，确保长期稳定运行。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

解决方案

#电力中心温湿度监控

#以太网协议

电力中心配电柜监测：SNMP协议读取以太网温湿度记录仪，告警规则配置

电力中心配电柜监测：SNMP协议读取以太网温湿度记录仪，告警规则配置

电力中心配电柜监测：SNMP协议读取以太网温湿度记录仪，告警规则配置

一、电力中心配电柜场景的特殊性

二、SNMP协议基础与记录仪MIB解析

2.1 SNMP版本选择

2.2 记录仪MIB库结构

2.3 加载MIB并读取数据

三、告警规则配置

3.1 告警层级设计

3.2 记录仪本地告警配置

3.3 SNMP Trap配置

3.4 联动规则引擎

四、典型坑

五、小结

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐