尧图网页设计 
简介本资源是一份面向Python开发者与能源领域技术人员的电力能耗分析实战项目文档聚焦工业制造、商业建筑等场景的精细化能耗管理与智能决策支持。内容覆盖从智能电表数据采集、MySQL数据库设计、pandas/Scikit-learn特征工程与多模型融合负荷预测、聚类分析、异常检测到FastAPI后端接口与Streamlit可视化前端的全流程实现兼顾技术深度与业务落地。资源为1个105KB的DOCX文档含完整目录结构项目背景与双碳政策契合点、四大技术层级架构图解、6类核心代码示例含数据预处理、时间序列建模、Matplotlib可视化等、4大典型应用领域说明及部署扩展建议便于读者快速掌握系统设计逻辑并开展本地实操或二次开发。目前已有72人学习下载适合具备Python基础的数据分析师、能源系统工程师及教学研究者参考复用。1. 为什么单靠一个模型做电力能耗分析总在“峰谷识别不准”“异常归因模糊”“领导问‘明天空调该开几台’答不上来”上反复翻车这不是数据不够多、不是Python没学好而是把电力能耗分析当成了“调个sklearn模型跑个准确率”的单点任务。真实产线里一台变压器的电流波动可能由设备启停、环境温湿度、生产排程三股力量同时拉扯某栋办公楼的月度用电突增既可能是新装了20台GPU服务器也可能是中央空调维保后能效下降了12%——单一模型根本分不清这是“量变”还是“质变”。本项目标题里那个被很多人跳过的词“多模型融合系统”才是破局关键。它不追求某个指标的绝对最优而是让LSTM抓时序依赖、XGBoost判设备级归因、孤立森林筛隐藏异常、规则引擎兜底业务逻辑最后用加权投票置信度校准输出可解释决策。适合正在做能源管理系统EMS落地、需要向运维/节能部门交付“能说清原因、能给出动作建议”的工程师也适合高校做综合能源课题、卡在“模型效果还行但业务方不认”的研究生。全文所有代码、数据库结构、GUI交互逻辑都围绕“让分析结果能进值班室大屏、能写进节能改造报告”这个硬需求展开。2. 数据采集层绕过SCADA协议黑匣子用Modbus TCP 本地缓存双通道拿下实时电表数据电力现场的数据源头往往比想象中脆弱老式电表只支持RS485 Modbus RTU新装智能终端走MQTT但网络不稳定而SCADA系统又常被厂商锁死API。硬啃OPC UA或定制驱动周期长、授权贵、后期维护难。我们选择一条更务实的路用Python构建轻量级采集代理以Modbus TCP为统一入口兼容物理转换器如RS485转TCP网关再叠加本地SQLite缓存兜底。这样即使网络中断30分钟数据也不丢重连后自动续传。2.1 用pymodbus构建健壮采集器超时、重试、断线重连全可控# collector/modbus_collector.py from pymodbus.client import ModbusTcpClient from pymodbus.exceptions import ModbusIOException, ConnectionException import time import logging class RobustModbusClient: def __init__(self, host, port502, timeout3, retries3): self.host host self.port port self.timeout timeout self.retries retries self.client None self._connect() def _connect(self): 带指数退避的连接避免雪崩重连 for i in range(self.retries): try: self.client ModbusTcpClient( hostself.host, portself.port, timeoutself.timeout, retry_on_emptyTrue, retries1 ) if self.client.connect(): logging.info(fModbus连接成功: {self.host}:{self.port}) return True except Exception as e: wait_time min(2 ** i, 30) # 最大等待30秒 logging.warning(f连接失败第{i1}次: {e}{wait_time}秒后重试...) time.sleep(wait_time) raise ConnectionError(fModbus连接重试{self.retries}次均失败) def read_holding_registers(self, address, count, slave1): 读取保持寄存器带异常捕获和重试 for i in range(self.retries): try: result self.client.read_holding_registers( addressaddress, countcount, slaveslave ) if not result.isError(): return result.registers else: logging.error(fModbus读取错误: {result}) except (ModbusIOException, ConnectionException) as e: logging.error(fModbus通信异常: {e}第{i1}次重试) if i self.retries - 1: time.sleep(1) self._reconnect_if_needed() return None def _reconnect_if_needed(self): 检测连接状态必要时重建 if not self.client or not self.client.is_socket_open(): logging.info(Modbus连接已断开尝试重建...) self.client.close() self._connect() # 使用示例读取电表地址40001-40004电压、电流、有功功率、功率因数 if __name__ __main__: client RobustModbusClient(192.168.1.100, port502) # 读取4个寄存器起始地址40001对应pymodbus的0x0000需减1 data client.read_holding_registers(address0, count4, slave1) if data: voltage data[0] / 10.0 # 假设寄存器值需除10得实际电压(V) current data[1] / 100.0 # 电流(A) power data[2] # 有功功率(W) pf data[3] / 1000.0 # 功率因数 print(f电压:{voltage:.1f}V, 电流:{current:.2f}A, 功率:{power}W, PF:{pf:.3f})逻辑说明与参数说明timeout3是关键——太短1s易被瞬时干扰误判断线太长5s导致采集周期拖垮实测3秒在工业现场平衡性最佳。retries3配合2**i指数退避避免网络抖动时大量并发重连压垮网关。address0对应Modbus标准地址40001这是硬件寄存器编号与软件索引的常见偏移必须查电表手册确认此处是血泪经验某品牌电表40001对应索引1而非0错一位导致所有数据错位。返回值data[0]/10.0的缩放因子scale factor必须从电表说明书获取不同厂家差异极大有/10、/100、甚至无缩放硬编码会翻车。2.2 SQLite本地缓存断网30分钟数据不丢重连后自动同步当Modbus采集线程因网络中断暂停内存中的数据会丢失。我们用SQLite做本地暂存每5秒写入一次结构极简-- db/energy_cache.db CREATE TABLE IF NOT EXISTS raw_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, device_id TEXT NOT NULL, voltage REAL, current REAL, active_power REAL, power_factor REAL, is_synced BOOLEAN DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_device_time ON raw_data(device_id, timestamp);采集主循环中插入缓存# collector/main_collector.py import sqlite3 from datetime import datetime def save_to_cache(conn, device_id, data_dict): 保存到SQLite缓存is_synced0表示待同步 cursor conn.cursor() cursor.execute( INSERT INTO raw_data (device_id, voltage, current, active_power, power_factor) VALUES (?, ?, ?, ?, ?) , (device_id, data_dict[voltage], data_dict[current], data_dict[active_power], data_dict[power_factor])) conn.commit() def sync_to_central_db(cache_conn, central_conn): 将未同步数据批量同步到中心数据库成功后标记is_synced1 cursor cache_conn.cursor() cursor.execute(SELECT * FROM raw_data WHERE is_synced 0 ORDER BY timestamp LIMIT 1000) records cursor.fetchall() if not records: return 0 # 批量插入中心库假设中心库是MySQL/PostgreSQL central_cursor central_conn.cursor() try: central_cursor.executemany( INSERT INTO energy_data (timestamp, device_id, voltage, current, active_power, power_factor) VALUES (?, ?, ?, ?, ?, ?) , [(r[1], r[2], r[3], r[4], r[5], r[6]) for r in records]) central_conn.commit() # 标记缓存中已同步 ids [r[0] for r in records] cursor.execute(fUPDATE raw_data SET is_synced 1 WHERE id IN ({,.join([?]*len(ids))}), ids) cache_conn.commit() return len(records) except Exception as e: logging.error(f同步失败: {e}) return 0 # 主采集循环简化 cache_conn sqlite3.connect(db/energy_cache.db) central_conn get_central_db_connection() # 实际连接MySQL等 while True: try: data modbus_client.read_all_sensors() if data: save_to_cache(cache_conn, METER_001, data) # 每30秒尝试同步一次 if time.time() % 30 1: synced_count sync_to_central_db(cache_conn, central_conn) logging.info(f同步{synced_count}条记录到中心库) except Exception as e: logging.error(f采集主循环异常: {e}) time.sleep(5)为什么选SQLite而非Redis或纯文件Redis虽快但断电丢数据不符合“断网不丢”要求CSV文件并发写入易损坏且无事务保障SQLite支持ACID、单文件部署、零配置PRAGMA journal_modeWAL开启后写入性能足够应付每5秒一次的采集节奏。关键技巧is_synced字段用布尔型而非时间戳避免同步失败时重复插入——这是某次凌晨3点同步失败导致数据库主键冲突的后悔药。3. 多模型融合层LSTM抓趋势、XGBoost定责任、孤立森林挖暗雷三模型投票不拼精度拼可解释性电力能耗分析最怕“黑匣子结论”。领导问“为什么7月用电涨了15%”模型回一句“LSTM预测误差2%”毫无价值。我们的融合策略核心是每个模型负责一个可解释维度最终决策是三个模型结论的加权交集而非简单平均。LSTM输出未来24小时负荷曲线趋势XGBoost输出各设备对异常时段的贡献度排序归因孤立森林标记疑似故障点异常。三者结论一致才触发告警否则进入人工复核队列。3.1 LSTM时序预测用滑动窗口构造特征避开“未来信息泄露”玄学陷阱很多教程直接用df.shift()生成标签导致训练时模型偷看了未来数据。我们严格按工业场景构造输入是过去1小时60个5分钟点的有功功率预测未来15分钟、30分钟、60分钟三个点。# models/lstm_predictor.py import numpy as np import pandas as pd from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, Dropout from sklearn.preprocessing import StandardScaler import joblib class PowerLSTM: def __init__(self, window_size60, pred_steps[3,6,12]): # 315min, 630min, 1260min self.window_size window_size self.pred_steps pred_steps self.scaler StandardScaler() self.model None def _create_sequences(self, data): 构造滑动窗口序列X: [batch, window, features], y: [batch, pred_steps] X, y [], [] for i in range(len(data) - self.window_size - max(self.pred_steps)): # 输入当前时刻往前window_size个点 x_window data[i:(i self.window_size)] # 输出未来pred_steps个点注意y的索引是iwindow_sizestep y_window [data[i self.window_size step] for step in self.pred_steps] X.append(x_window) y.append(y_window) return np.array(X), np.array(y) def train(self, df_power, model_pathmodels/lstm_power.h5): df_power: pd.Series, indexdatetime, valuesactive_power(W) # 仅用有功功率一维序列简化版实际可加温度、湿度 power_series df_power.values.reshape(-1, 1) scaled_power self.scaler.fit_transform(power_series).flatten() X, y self._create_sequences(scaled_power) # Reshape for LSTM: (samples, timesteps, features) X X.reshape((X.shape[0], X.shape[1], 1)) self.model Sequential([ LSTM(50, return_sequencesTrue, input_shape(X.shape[1], 1)), Dropout(0.2), LSTM(50, return_sequencesFalse), Dropout(0.2), Dense(25), Dense(len(self.pred_steps)) # 输出3个预测点 ]) self.model.compile(optimizeradam, lossmse) self.model.fit(X, y, epochs50, batch_size32, verbose0) self.model.save(model_path) joblib.dump(self.scaler, models/lstm_scaler.pkl) print(fLSTM模型训练完成保存至{model_path}) def predict_next(self, recent_power_array): recent_power_array: np.array, shape(60,), 最近60个5分钟点的功率值 scaled_input self.scaler.transform(recent_power_array.reshape(-1,1)).flatten() X_pred scaled_input.reshape(1, self.window_size, 1) pred_scaled self.model.predict(X_pred) # 反标准化 pred_actual self.scaler.inverse_transform(pred_scaled.reshape(-1,1)).flatten() return pred_actual # array of length 3: [15min, 30min, 60min] # 使用示例训练 # df pd.read_sql(SELECT timestamp, active_power FROM energy_data WHERE device_idMETER_001 ORDER BY timestamp, conn) # lstm PowerLSTM() # lstm.train(df.set_index(timestamp)[active_power])参数说明与避坑window_size60对应1小时历史数据经测试小于3030分钟时峰谷识别率骤降大于1202小时则训练慢且边际收益低pred_steps[3,6,12]严格对应业务需求15分钟预警够运维人员响应、30分钟调度调整空调设定、60分钟计划通知生产排程致命坑_create_sequences中y_window的索引i self.window_size step必须确保不越界否则训练报错或数据错乱——某次调试因max(self.pred_steps)计算错误导致最后一批数据y全为0模型学会“永远预测0”上线后告警全失效。3.2 XGBoost设备归因用SHAP值代替特征重要性回答“空调还是服务器在耗电”单纯看XGBoost的feature_importance_只能知道“电压”比“电流”重要但无法回答“7月23日14:00-15:00的异常高耗电是3楼空调还是数据中心机房导致的”。我们用SHAPSHapley Additive exPlanations计算每个设备在异常时段的贡献值。# models/xgb_attributor.py import xgboost as xgb import shap import pandas as pd import numpy as np class DeviceAttributor: def __init__(self, feature_colsNone): self.feature_cols feature_cols or [voltage, current, temp_indoor, temp_outdoor, hour, is_weekend] self.model None self.explainer None def prepare_features(self, df): 构造归因特征每个设备一行包含其自身及关联环境特征 # 假设df有 device_id, timestamp, voltage, current, temp_indoor, temp_outdoor df_feat df.copy() df_feat[hour] pd.to_datetime(df_feat[timestamp]).dt.hour df_feat[is_weekend] (pd.to_datetime(df_feat[timestamp]).dt.dayofweek 5).astype(int) return df_feat[self.feature_cols] def train(self, X_train, y_train, model_pathmodels/xgb_attributor.json): y_train是二分类标签0正常1异常时段由孤立森林或阈值判定 self.model xgb.XGBClassifier( n_estimators200, max_depth6, learning_rate0.1, subsample0.8, colsample_bytree0.8, random_state42 ) self.model.fit(X_train, y_train) # 训练SHAP解释器用KernelExplainer更鲁棒但慢TreeExplainer快且适配XGB self.explainer shap.TreeExplainer(self.model) self.model.save_model(model_path) print(fXGBoost归因模型训练完成) def explain_anomaly(self, X_single): 对单个样本如异常时段均值计算各特征SHAP值 if self.explainer is None: raise ValueError(请先训练模型) shap_values self.explainer.shap_values(X_single) # shap_values是二维数组取正类异常的shap值 shap_df pd.DataFrame({ feature: self.feature_cols, shap_value: shap_values[0] if isinstance(shap_values, list) else shap_values[0] }).sort_values(shap_value, keyabs, ascendingFalse) return shap_df # 使用示例分析某异常时段 # anomaly_df pd.read_sql(SELECT * FROM energy_data WHERE timestamp BETWEEN 2024-07-23 14:00 AND 2024-07-23 15:00, conn) # X_anomaly attributor.prepare_features(anomaly_df) # X_mean X_anomaly.mean().values.reshape(1, -1) # 取时段均值作为代表样本 # shap_result attributor.explain_anomaly(X_mean) # print(shap_result.head(3)) # 输出贡献度Top3的特征如 temp_indoor: 0.42, current: 0.38...为什么SHAP比feature_importance有用feature_importance_说“电压重要”SHAP说“在14:00这个时刻室内温度比均值高3℃导致模型判断为异常的贡献度是0.42”SHAP值可正可负正值表示推高异常概率负值表示抑制——比如“周末”特征为-0.25说明周末本应耗电少但此时仍高问题更严重血泪经验shap_values[0]的索引取决于XGBoost输出格式二分类时shap_values是list of 2 arrays必须取[0]正类取错导致归因方向全反。3.3 孤立森林异常检测用contamination参数直连业务告别“调参像玄学”孤立森林Isolation Forest的contamination参数常被当成超参调优其实它就是业务定义的“异常比例”。我们直接设为0.022%因为电力运维约定每天允许2%的时间段出现非计划性波动如设备启停超出即需人工核查。# models/isoforest_detector.py from sklearn.ensemble import IsolationForest import pandas as pd import numpy as np class PowerAnomalyDetector: def __init__(self, contamination0.02, n_estimators100): self.contamination contamination self.n_estimators n_estimators self.model IsolationForest( contaminationself.contamination, n_estimatorsself.n_estimators, max_samplesauto, random_state42, n_jobs-1 ) self.feature_cols [active_power, voltage, current, power_factor] def prepare_features(self, df): 构造异常检测特征功率为主辅以电气质量参数 return df[self.feature_cols].values def train(self, df, model_pathmodels/isoforest.pkl): X self.prepare_features(df) self.model.fit(X) joblib.dump(self.model, model_path) print(f孤立森林模型训练完成contamination{self.contamination}) def detect_batch(self, df): 批量检测返回异常标记和异常得分 X self.prepare_features(df) # predict: -1为异常1为正常decision_function返回异常程度越负越异常 y_pred self.model.predict(X) anomaly_scores self.model.decision_function(X) result_df df.copy() result_df[is_anomaly] (y_pred -1).astype(int) result_df[anomaly_score] anomaly_scores return result_df # 使用示例 # df_history pd.read_sql(SELECT * FROM energy_data WHERE device_idMETER_001 AND timestamp 2024-07-01, conn) # detector PowerAnomalyDetector(contamination0.02) # detector.train(df_history) # df_today pd.read_sql(SELECT * FROM energy_data WHERE device_idMETER_001 AND date(timestamp)2024-07-23, conn) # df_labeled detector.detect_batch(df_today) # anomalies df_labeled[df_labeled[is_anomaly]1] # print(f今日发现{len(anomalies)}个异常点最高分:{anomalies[anomaly_score].min():.3f})contamination参数的业务直译设为0.02 ≠ “模型认为2%数据是异常”而是“我业务上定义每天最多容忍2%的异常时段超出就报警”这样设置后模型自动调整分割阈值无需手动调max_samples或n_estimators——某次为追求“更高召回率”把contamination调到0.05结果每天告警30次运维直接拒收anomaly_score越负表示越异常我们用它排序只推送Top5给值班员避免信息过载。4. 决策支持层规则引擎兜底、GUI可视化交互、数据库联动让分析结果变成可执行动作模型输出只是中间产物真正的价值在于“下一步该做什么”。我们设计三层决策流模型层输出LSTM趋势/XGBoost归因/孤立森林异常→ 规则引擎翻译成业务语言 → GUI界面呈现可点击动作。例如当LSTM预测15分钟后负荷将超阈值110%且XGBoost归因显示空调贡献度60%规则引擎立即触发“建议关闭3楼东侧空调”动作并在GUI上高亮对应空调设备图标。4.1 规则引擎用Drools思想实现Python轻量级规则库不用引入复杂规则引擎框架用Python字典函数映射实现可维护规则# decision/rules_engine.py from datetime import datetime, timedelta import pandas as pd class EnergyRuleEngine: def __init__(self): # 规则库key规则IDvalue规则函数 self.rules { high_load_warning: self._rule_high_load_warning, ac_overload_response: self._rule_ac_overload_response, low_power_factor_alert: self._rule_low_power_factor_alert, } def _rule_high_load_warning(self, context): 高负荷预警规则LSTM预测15min后超阈值且XGBoost归因空调60% if not context.get(lstm_prediction) or not context.get(xgb_shap): return None pred_15min context[lstm_prediction][0] # 第一个预测值15min后 ac_contribution context[xgb_shap].query(feature current)[shap_value].iloc[0] threshold context.get(load_threshold, 8000) # W if pred_15min threshold * 1.1 and ac_contribution 0.6: return { action: recommend_ac_shutdown, target_devices: [AC_3F_EAST, AC_3F_WEST], reason: f预测15分钟后负荷达{pred_15min:.0f}W超阈值10%空调电流贡献度{ac_contribution:.1%}, priority: high } return None def _rule_ac_overload_response(self, context): 空调过载响应孤立森林标记异常且该时段空调电流突增50% if not context.get(isoforest_result) or not context.get(raw_data): return None anomaly_df context[isoforest_result] raw_df context[raw_data] # 找出最近一个异常点 last_anomaly anomaly_df[anomaly_df[is_anomaly]1].tail(1) if last_anomaly.empty: return None # 计算该异常点前1小时电流均值与当前电流对比 current_now last_anomaly[current].iloc[0] time_anomaly last_anomaly[timestamp].iloc[0] hour_ago time_anomaly - timedelta(hours1) recent_currents raw_df[ (raw_df[timestamp] hour_ago) (raw_df[timestamp] time_anomaly) ][current] if len(recent_currents) 0: return None current_avg recent_currents.mean() if current_now current_avg * 1.5: return { action: alert_ac_maintenance, target_devices: [AC_3F_EAST], reason: f空调电流突增至{current_now:.1f}A较1小时前均值{current_avg:.1f}A升50%, priority: critical } return None def execute_all_rules(self, context): 执行所有规则返回最高优先级动作 actions [] for rule_id, rule_func in self.rules.items(): try: action rule_func(context) if action: actions.append(action) except Exception as e: print(f规则{rule_id}执行异常: {e}) if not actions: return None # 按priority排序critical high medium priority_map {critical: 3, high: 2, medium: 1} actions.sort(keylambda x: priority_map.get(x[priority], 0), reverseTrue) return actions[0] # 返回最高优先级动作 # 使用示例 # context { # lstm_prediction: [8500, 8700, 8900], # 15/30/60min预测 # xgb_shap: shap_result, # XGBoost的SHAP分析结果DataFrame # isoforest_result: df_labeled, # 孤立森林标注结果 # raw_data: df_today, # 当日原始数据 # load_threshold: 8000 # } # engine EnergyRuleEngine() # action engine.execute_all_rules(context) # if action: # print(f触发动作: {action[action]}, 目标设备: {action[target_devices]})规则引擎设计哲学每个规则函数独立便于单元测试和业务方审核如让节能工程师确认“空调贡献度60%才建议关机”是否合理context字典传递所有上游模型结果解耦模型与规则模型升级不影响规则逻辑关键技巧execute_all_rules返回最高优先级动作而非全部动作——避免GUI同时弹出10个窗口这是某次测试时运维抱怨“告警比微信消息还多”的直接改进。4.2 GUI设计用PyQt5构建生产级监控界面重点在“一键导出报告”和“设备图谱联动”GUI不是炫技核心功能就两个实时看板趋势异常点 一键生成PDF节能报告。我们放弃Web方案部署复杂、离线不可用用PyQt5构建桌面应用所有图表用Matplotlib嵌入设备拓扑用QGraphicsScene绘制。# gui/main_window.py import sys from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QTabWidget, QLabel, QPushButton, QComboBox, QGroupBox, QGridLayout, QFileDialog) from PyQt5.QtCore import Qt, QTimer import matplotlib.pyplot as plt from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas import pandas as pd from reportlab.pdfgen import canvas from reportlab.lib.pagesizes import A4 from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Image from reportlab.lib.styles import getSampleStyleSheet class EnergyMonitorGUI(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle(电力能耗智能分析系统 v1.0) self.setGeometry(100, 100, 1400, 800) # 主布局 central_widget QWidget() self.setCentralWidget(central_widget) main_layout QVBoxLayout(central_widget) # 顶部控制栏 control_layout QHBoxLayout() self.device_combo QComboBox() self.device_combo.addItems([METER_001, METER_002, TRANSFORMER_A]) self.date_picker QLabel(2024-07-23) self.refresh_btn QPushButton(刷新数据) self.refresh_btn.clicked.connect(self.load_data) control_layout.addWidget(QLabel(设备:)) control_layout.addWidget(self.device_combo) control_layout.addWidget(QLabel(日期:)) control_layout.addWidget(self.date_picker) control_layout.addWidget(self.refresh_btn) control_layout.addStretch() main_layout.addLayout(control_layout) # 选项卡 self.tabs QTabWidget() main_layout.addWidget(self.tabs) # 实时趋势页 self.trend_tab QWidget() self.trend_layout QVBoxLayout(self.trend_tab) self.trend_canvas MatplotlibCanvas() self.trend_layout.addWidget(self.trend_canvas) self.tabs.addTab(self.trend_tab, 实时趋势) # 异常分析页 self.anomaly_tab QWidget() self.anomaly_layout QVBoxLayout(self.anomaly_tab) self.anomaly_canvas MatplotlibCanvas() self.anomaly_layout.addWidget(self.anomaly_canvas) self.export_btn QPushButton(导出今日分析报告 (PDF)) self.export_btn.clicked.connect(self.export_report) self.anomaly_layout.addWidget(self.export_btn) self.tabs.addTab(self.anomaly_tab, 异常分析) # 加载初始数据 self.load_data() # 定时刷新每60秒 self.timer QTimer() self.timer.timeout.connect(self.load_data) self.timer.start(60000) def load_data(self): 从数据库加载最新数据并更新图表 device_id self.device_combo.currentText() # 伪代码实际从MySQL查询 # df_trend pd.read_sql(fSELECT * FROM energy_data WHERE device_id{device_id} AND timestamp NOW() - INTERVAL 24 HOUR, conn) # df_anomaly pd.read_sql(fSELECT * FROM energy_data WHERE device_id{device_id} AND is_anomaly1 AND date(timestamp)CURDATE(), conn) # 更新趋势图模拟数据 self.trend_canvas.plot_trend([i for i in range(288)], [1000500*i%100 for i in range(288)]) # 288个5分钟点 # 更新异常图模拟数据 self.anomaly_canvas.plot_anomaly([10, 50, 120, 200], [1200, 1500, 1800, 2100]) # 异常点位置和值 def export_report(self): 一键导出PDF报告 device_id self.device_combo.currentText() filename, _ QFileDialog.getSaveFileName( self, 保存报告, f{device_id}_energy_report_{self.date_picker.text()}.pdf, PDF Files (*.pdf) ) if not filename: return # 构建PDF内容 doc SimpleDocTemplate(filename, pagesizeA4) styles getSampleStyleSheet() story [] story.append(Paragraph(f设备 {device_id} 能耗分析报告, styles[Title])) story.append(Spacer(1, 12)) story.append(Paragraph(f日期: {self.date_picker.text()}, styles[Normal])) story.append(Spacer(1, 12)) # 添加趋势图需先保存为图片 trend_img_path temp_trend.png self.trend_canvas.figure.savefig(trend_img_path, dpi150, bbox_inchestight) story.append(Image(trend_img_path, width400, height250)) story.append(Spacer(1, 12)) # 添加关键结论从规则引擎获取 # action rule_engine.execute_all_rules(context) # 实际调用 # if action: # story.append(Paragraph(f【关键建议】{action[reason]}, styles[Heading2])) # story.append(Paragraph(f执行动作: {action[action]}, styles[Normal])) doc.build(story) print(f报告已导出: {filename}) class MatplotlibCanvas(FigureCanvas): def __init__( p a hrefhttps://download.csdn.net/download/xiaoxingkongyuxi/90437377 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p 电力能耗多模型融合分析系统：LSTM+XGBoost+孤立森林实战
简介本资源是一份面向Python开发者与能源领域技术人员的电力能耗分析实战项目文档聚焦工业制造、商业建筑等场景的精细化能耗管理与智能决策支持。内容覆盖从智能电表数据采集、MySQL数据库设计、pandas/Scikit-learn特征工程与多模型融合负荷预测、聚类分析、异常检测到FastAPI后端接口与Streamlit可视化前端的全流程实现兼顾技术深度与业务落地。资源为1个105KB的DOCX文档含完整目录结构项目背景与双碳政策契合点、四大技术层级架构图解、6类核心代码示例含数据预处理、时间序列建模、Matplotlib可视化等、4大典型应用领域说明及部署扩展建议便于读者快速掌握系统设计逻辑并开展本地实操或二次开发。目前已有72人学习下载适合具备Python基础的数据分析师、能源系统工程师及教学研究者参考复用。1. 为什么单靠一个模型做电力能耗分析总在“峰谷识别不准”“异常归因模糊”“领导问‘明天空调该开几台’答不上来”上反复翻车这不是数据不够多、不是Python没学好而是把电力能耗分析当成了“调个sklearn模型跑个准确率”的单点任务。真实产线里一台变压器的电流波动可能由设备启停、环境温湿度、生产排程三股力量同时拉扯某栋办公楼的月度用电突增既可能是新装了20台GPU服务器也可能是中央空调维保后能效下降了12%——单一模型根本分不清这是“量变”还是“质变”。本项目标题里那个被很多人跳过的词“多模型融合系统”才是破局关键。它不追求某个指标的绝对最优而是让LSTM抓时序依赖、XGBoost判设备级归因、孤立森林筛隐藏异常、规则引擎兜底业务逻辑最后用加权投票置信度校准输出可解释决策。适合正在做能源管理系统EMS落地、需要向运维/节能部门交付“能说清原因、能给出动作建议”的工程师也适合高校做综合能源课题、卡在“模型效果还行但业务方不认”的研究生。全文所有代码、数据库结构、GUI交互逻辑都围绕“让分析结果能进值班室大屏、能写进节能改造报告”这个硬需求展开。2. 数据采集层绕过SCADA协议黑匣子用Modbus TCP 本地缓存双通道拿下实时电表数据电力现场的数据源头往往比想象中脆弱老式电表只支持RS485 Modbus RTU新装智能终端走MQTT但网络不稳定而SCADA系统又常被厂商锁死API。硬啃OPC UA或定制驱动周期长、授权贵、后期维护难。我们选择一条更务实的路用Python构建轻量级采集代理以Modbus TCP为统一入口兼容物理转换器如RS485转TCP网关再叠加本地SQLite缓存兜底。这样即使网络中断30分钟数据也不丢重连后自动续传。2.1 用pymodbus构建健壮采集器超时、重试、断线重连全可控# collector/modbus_collector.py from pymodbus.client import ModbusTcpClient from pymodbus.exceptions import ModbusIOException, ConnectionException import time import logging class RobustModbusClient: def __init__(self, host, port502, timeout3, retries3): self.host host self.port port self.timeout timeout self.retries retries self.client None self._connect() def _connect(self): 带指数退避的连接避免雪崩重连 for i in range(self.retries): try: self.client ModbusTcpClient( hostself.host, portself.port, timeoutself.timeout, retry_on_emptyTrue, retries1 ) if self.client.connect(): logging.info(fModbus连接成功: {self.host}:{self.port}) return True except Exception as e: wait_time min(2 ** i, 30) # 最大等待30秒 logging.warning(f连接失败第{i1}次: {e}{wait_time}秒后重试...) time.sleep(wait_time) raise ConnectionError(fModbus连接重试{self.retries}次均失败) def read_holding_registers(self, address, count, slave1): 读取保持寄存器带异常捕获和重试 for i in range(self.retries): try: result self.client.read_holding_registers( addressaddress, countcount, slaveslave ) if not result.isError(): return result.registers else: logging.error(fModbus读取错误: {result}) except (ModbusIOException, ConnectionException) as e: logging.error(fModbus通信异常: {e}第{i1}次重试) if i self.retries - 1: time.sleep(1) self._reconnect_if_needed() return None def _reconnect_if_needed(self): 检测连接状态必要时重建 if not self.client or not self.client.is_socket_open(): logging.info(Modbus连接已断开尝试重建...) self.client.close() self._connect() # 使用示例读取电表地址40001-40004电压、电流、有功功率、功率因数 if __name__ __main__: client RobustModbusClient(192.168.1.100, port502) # 读取4个寄存器起始地址40001对应pymodbus的0x0000需减1 data client.read_holding_registers(address0, count4, slave1) if data: voltage data[0] / 10.0 # 假设寄存器值需除10得实际电压(V) current data[1] / 100.0 # 电流(A) power data[2] # 有功功率(W) pf data[3] / 1000.0 # 功率因数 print(f电压:{voltage:.1f}V, 电流:{current:.2f}A, 功率:{power}W, PF:{pf:.3f})逻辑说明与参数说明timeout3是关键——太短1s易被瞬时干扰误判断线太长5s导致采集周期拖垮实测3秒在工业现场平衡性最佳。retries3配合2**i指数退避避免网络抖动时大量并发重连压垮网关。address0对应Modbus标准地址40001这是硬件寄存器编号与软件索引的常见偏移必须查电表手册确认此处是血泪经验某品牌电表40001对应索引1而非0错一位导致所有数据错位。返回值data[0]/10.0的缩放因子scale factor必须从电表说明书获取不同厂家差异极大有/10、/100、甚至无缩放硬编码会翻车。2.2 SQLite本地缓存断网30分钟数据不丢重连后自动同步当Modbus采集线程因网络中断暂停内存中的数据会丢失。我们用SQLite做本地暂存每5秒写入一次结构极简-- db/energy_cache.db CREATE TABLE IF NOT EXISTS raw_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, device_id TEXT NOT NULL, voltage REAL, current REAL, active_power REAL, power_factor REAL, is_synced BOOLEAN DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_device_time ON raw_data(device_id, timestamp);采集主循环中插入缓存# collector/main_collector.py import sqlite3 from datetime import datetime def save_to_cache(conn, device_id, data_dict): 保存到SQLite缓存is_synced0表示待同步 cursor conn.cursor() cursor.execute( INSERT INTO raw_data (device_id, voltage, current, active_power, power_factor) VALUES (?, ?, ?, ?, ?) , (device_id, data_dict[voltage], data_dict[current], data_dict[active_power], data_dict[power_factor])) conn.commit() def sync_to_central_db(cache_conn, central_conn): 将未同步数据批量同步到中心数据库成功后标记is_synced1 cursor cache_conn.cursor() cursor.execute(SELECT * FROM raw_data WHERE is_synced 0 ORDER BY timestamp LIMIT 1000) records cursor.fetchall() if not records: return 0 # 批量插入中心库假设中心库是MySQL/PostgreSQL central_cursor central_conn.cursor() try: central_cursor.executemany( INSERT INTO energy_data (timestamp, device_id, voltage, current, active_power, power_factor) VALUES (?, ?, ?, ?, ?, ?) , [(r[1], r[2], r[3], r[4], r[5], r[6]) for r in records]) central_conn.commit() # 标记缓存中已同步 ids [r[0] for r in records] cursor.execute(fUPDATE raw_data SET is_synced 1 WHERE id IN ({,.join([?]*len(ids))}), ids) cache_conn.commit() return len(records) except Exception as e: logging.error(f同步失败: {e}) return 0 # 主采集循环简化 cache_conn sqlite3.connect(db/energy_cache.db) central_conn get_central_db_connection() # 实际连接MySQL等 while True: try: data modbus_client.read_all_sensors() if data: save_to_cache(cache_conn, METER_001, data) # 每30秒尝试同步一次 if time.time() % 30 1: synced_count sync_to_central_db(cache_conn, central_conn) logging.info(f同步{synced_count}条记录到中心库) except Exception as e: logging.error(f采集主循环异常: {e}) time.sleep(5)为什么选SQLite而非Redis或纯文件Redis虽快但断电丢数据不符合“断网不丢”要求CSV文件并发写入易损坏且无事务保障SQLite支持ACID、单文件部署、零配置PRAGMA journal_modeWAL开启后写入性能足够应付每5秒一次的采集节奏。关键技巧is_synced字段用布尔型而非时间戳避免同步失败时重复插入——这是某次凌晨3点同步失败导致数据库主键冲突的后悔药。3. 多模型融合层LSTM抓趋势、XGBoost定责任、孤立森林挖暗雷三模型投票不拼精度拼可解释性电力能耗分析最怕“黑匣子结论”。领导问“为什么7月用电涨了15%”模型回一句“LSTM预测误差2%”毫无价值。我们的融合策略核心是每个模型负责一个可解释维度最终决策是三个模型结论的加权交集而非简单平均。LSTM输出未来24小时负荷曲线趋势XGBoost输出各设备对异常时段的贡献度排序归因孤立森林标记疑似故障点异常。三者结论一致才触发告警否则进入人工复核队列。3.1 LSTM时序预测用滑动窗口构造特征避开“未来信息泄露”玄学陷阱很多教程直接用df.shift()生成标签导致训练时模型偷看了未来数据。我们严格按工业场景构造输入是过去1小时60个5分钟点的有功功率预测未来15分钟、30分钟、60分钟三个点。# models/lstm_predictor.py import numpy as np import pandas as pd from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, Dropout from sklearn.preprocessing import StandardScaler import joblib class PowerLSTM: def __init__(self, window_size60, pred_steps[3,6,12]): # 315min, 630min, 1260min self.window_size window_size self.pred_steps pred_steps self.scaler StandardScaler() self.model None def _create_sequences(self, data): 构造滑动窗口序列X: [batch, window, features], y: [batch, pred_steps] X, y [], [] for i in range(len(data) - self.window_size - max(self.pred_steps)): # 输入当前时刻往前window_size个点 x_window data[i:(i self.window_size)] # 输出未来pred_steps个点注意y的索引是iwindow_sizestep y_window [data[i self.window_size step] for step in self.pred_steps] X.append(x_window) y.append(y_window) return np.array(X), np.array(y) def train(self, df_power, model_pathmodels/lstm_power.h5): df_power: pd.Series, indexdatetime, valuesactive_power(W) # 仅用有功功率一维序列简化版实际可加温度、湿度 power_series df_power.values.reshape(-1, 1) scaled_power self.scaler.fit_transform(power_series).flatten() X, y self._create_sequences(scaled_power) # Reshape for LSTM: (samples, timesteps, features) X X.reshape((X.shape[0], X.shape[1], 1)) self.model Sequential([ LSTM(50, return_sequencesTrue, input_shape(X.shape[1], 1)), Dropout(0.2), LSTM(50, return_sequencesFalse), Dropout(0.2), Dense(25), Dense(len(self.pred_steps)) # 输出3个预测点 ]) self.model.compile(optimizeradam, lossmse) self.model.fit(X, y, epochs50, batch_size32, verbose0) self.model.save(model_path) joblib.dump(self.scaler, models/lstm_scaler.pkl) print(fLSTM模型训练完成保存至{model_path}) def predict_next(self, recent_power_array): recent_power_array: np.array, shape(60,), 最近60个5分钟点的功率值 scaled_input self.scaler.transform(recent_power_array.reshape(-1,1)).flatten() X_pred scaled_input.reshape(1, self.window_size, 1) pred_scaled self.model.predict(X_pred) # 反标准化 pred_actual self.scaler.inverse_transform(pred_scaled.reshape(-1,1)).flatten() return pred_actual # array of length 3: [15min, 30min, 60min] # 使用示例训练 # df pd.read_sql(SELECT timestamp, active_power FROM energy_data WHERE device_idMETER_001 ORDER BY timestamp, conn) # lstm PowerLSTM() # lstm.train(df.set_index(timestamp)[active_power])参数说明与避坑window_size60对应1小时历史数据经测试小于3030分钟时峰谷识别率骤降大于1202小时则训练慢且边际收益低pred_steps[3,6,12]严格对应业务需求15分钟预警够运维人员响应、30分钟调度调整空调设定、60分钟计划通知生产排程致命坑_create_sequences中y_window的索引i self.window_size step必须确保不越界否则训练报错或数据错乱——某次调试因max(self.pred_steps)计算错误导致最后一批数据y全为0模型学会“永远预测0”上线后告警全失效。3.2 XGBoost设备归因用SHAP值代替特征重要性回答“空调还是服务器在耗电”单纯看XGBoost的feature_importance_只能知道“电压”比“电流”重要但无法回答“7月23日14:00-15:00的异常高耗电是3楼空调还是数据中心机房导致的”。我们用SHAPSHapley Additive exPlanations计算每个设备在异常时段的贡献值。# models/xgb_attributor.py import xgboost as xgb import shap import pandas as pd import numpy as np class DeviceAttributor: def __init__(self, feature_colsNone): self.feature_cols feature_cols or [voltage, current, temp_indoor, temp_outdoor, hour, is_weekend] self.model None self.explainer None def prepare_features(self, df): 构造归因特征每个设备一行包含其自身及关联环境特征 # 假设df有 device_id, timestamp, voltage, current, temp_indoor, temp_outdoor df_feat df.copy() df_feat[hour] pd.to_datetime(df_feat[timestamp]).dt.hour df_feat[is_weekend] (pd.to_datetime(df_feat[timestamp]).dt.dayofweek 5).astype(int) return df_feat[self.feature_cols] def train(self, X_train, y_train, model_pathmodels/xgb_attributor.json): y_train是二分类标签0正常1异常时段由孤立森林或阈值判定 self.model xgb.XGBClassifier( n_estimators200, max_depth6, learning_rate0.1, subsample0.8, colsample_bytree0.8, random_state42 ) self.model.fit(X_train, y_train) # 训练SHAP解释器用KernelExplainer更鲁棒但慢TreeExplainer快且适配XGB self.explainer shap.TreeExplainer(self.model) self.model.save_model(model_path) print(fXGBoost归因模型训练完成) def explain_anomaly(self, X_single): 对单个样本如异常时段均值计算各特征SHAP值 if self.explainer is None: raise ValueError(请先训练模型) shap_values self.explainer.shap_values(X_single) # shap_values是二维数组取正类异常的shap值 shap_df pd.DataFrame({ feature: self.feature_cols, shap_value: shap_values[0] if isinstance(shap_values, list) else shap_values[0] }).sort_values(shap_value, keyabs, ascendingFalse) return shap_df # 使用示例分析某异常时段 # anomaly_df pd.read_sql(SELECT * FROM energy_data WHERE timestamp BETWEEN 2024-07-23 14:00 AND 2024-07-23 15:00, conn) # X_anomaly attributor.prepare_features(anomaly_df) # X_mean X_anomaly.mean().values.reshape(1, -1) # 取时段均值作为代表样本 # shap_result attributor.explain_anomaly(X_mean) # print(shap_result.head(3)) # 输出贡献度Top3的特征如 temp_indoor: 0.42, current: 0.38...为什么SHAP比feature_importance有用feature_importance_说“电压重要”SHAP说“在14:00这个时刻室内温度比均值高3℃导致模型判断为异常的贡献度是0.42”SHAP值可正可负正值表示推高异常概率负值表示抑制——比如“周末”特征为-0.25说明周末本应耗电少但此时仍高问题更严重血泪经验shap_values[0]的索引取决于XGBoost输出格式二分类时shap_values是list of 2 arrays必须取[0]正类取错导致归因方向全反。3.3 孤立森林异常检测用contamination参数直连业务告别“调参像玄学”孤立森林Isolation Forest的contamination参数常被当成超参调优其实它就是业务定义的“异常比例”。我们直接设为0.022%因为电力运维约定每天允许2%的时间段出现非计划性波动如设备启停超出即需人工核查。# models/isoforest_detector.py from sklearn.ensemble import IsolationForest import pandas as pd import numpy as np class PowerAnomalyDetector: def __init__(self, contamination0.02, n_estimators100): self.contamination contamination self.n_estimators n_estimators self.model IsolationForest( contaminationself.contamination, n_estimatorsself.n_estimators, max_samplesauto, random_state42, n_jobs-1 ) self.feature_cols [active_power, voltage, current, power_factor] def prepare_features(self, df): 构造异常检测特征功率为主辅以电气质量参数 return df[self.feature_cols].values def train(self, df, model_pathmodels/isoforest.pkl): X self.prepare_features(df) self.model.fit(X) joblib.dump(self.model, model_path) print(f孤立森林模型训练完成contamination{self.contamination}) def detect_batch(self, df): 批量检测返回异常标记和异常得分 X self.prepare_features(df) # predict: -1为异常1为正常decision_function返回异常程度越负越异常 y_pred self.model.predict(X) anomaly_scores self.model.decision_function(X) result_df df.copy() result_df[is_anomaly] (y_pred -1).astype(int) result_df[anomaly_score] anomaly_scores return result_df # 使用示例 # df_history pd.read_sql(SELECT * FROM energy_data WHERE device_idMETER_001 AND timestamp 2024-07-01, conn) # detector PowerAnomalyDetector(contamination0.02) # detector.train(df_history) # df_today pd.read_sql(SELECT * FROM energy_data WHERE device_idMETER_001 AND date(timestamp)2024-07-23, conn) # df_labeled detector.detect_batch(df_today) # anomalies df_labeled[df_labeled[is_anomaly]1] # print(f今日发现{len(anomalies)}个异常点最高分:{anomalies[anomaly_score].min():.3f})contamination参数的业务直译设为0.02 ≠ “模型认为2%数据是异常”而是“我业务上定义每天最多容忍2%的异常时段超出就报警”这样设置后模型自动调整分割阈值无需手动调max_samples或n_estimators——某次为追求“更高召回率”把contamination调到0.05结果每天告警30次运维直接拒收anomaly_score越负表示越异常我们用它排序只推送Top5给值班员避免信息过载。4. 决策支持层规则引擎兜底、GUI可视化交互、数据库联动让分析结果变成可执行动作模型输出只是中间产物真正的价值在于“下一步该做什么”。我们设计三层决策流模型层输出LSTM趋势/XGBoost归因/孤立森林异常→ 规则引擎翻译成业务语言 → GUI界面呈现可点击动作。例如当LSTM预测15分钟后负荷将超阈值110%且XGBoost归因显示空调贡献度60%规则引擎立即触发“建议关闭3楼东侧空调”动作并在GUI上高亮对应空调设备图标。4.1 规则引擎用Drools思想实现Python轻量级规则库不用引入复杂规则引擎框架用Python字典函数映射实现可维护规则# decision/rules_engine.py from datetime import datetime, timedelta import pandas as pd class EnergyRuleEngine: def __init__(self): # 规则库key规则IDvalue规则函数 self.rules { high_load_warning: self._rule_high_load_warning, ac_overload_response: self._rule_ac_overload_response, low_power_factor_alert: self._rule_low_power_factor_alert, } def _rule_high_load_warning(self, context): 高负荷预警规则LSTM预测15min后超阈值且XGBoost归因空调60% if not context.get(lstm_prediction) or not context.get(xgb_shap): return None pred_15min context[lstm_prediction][0] # 第一个预测值15min后 ac_contribution context[xgb_shap].query(feature current)[shap_value].iloc[0] threshold context.get(load_threshold, 8000) # W if pred_15min threshold * 1.1 and ac_contribution 0.6: return { action: recommend_ac_shutdown, target_devices: [AC_3F_EAST, AC_3F_WEST], reason: f预测15分钟后负荷达{pred_15min:.0f}W超阈值10%空调电流贡献度{ac_contribution:.1%}, priority: high } return None def _rule_ac_overload_response(self, context): 空调过载响应孤立森林标记异常且该时段空调电流突增50% if not context.get(isoforest_result) or not context.get(raw_data): return None anomaly_df context[isoforest_result] raw_df context[raw_data] # 找出最近一个异常点 last_anomaly anomaly_df[anomaly_df[is_anomaly]1].tail(1) if last_anomaly.empty: return None # 计算该异常点前1小时电流均值与当前电流对比 current_now last_anomaly[current].iloc[0] time_anomaly last_anomaly[timestamp].iloc[0] hour_ago time_anomaly - timedelta(hours1) recent_currents raw_df[ (raw_df[timestamp] hour_ago) (raw_df[timestamp] time_anomaly) ][current] if len(recent_currents) 0: return None current_avg recent_currents.mean() if current_now current_avg * 1.5: return { action: alert_ac_maintenance, target_devices: [AC_3F_EAST], reason: f空调电流突增至{current_now:.1f}A较1小时前均值{current_avg:.1f}A升50%, priority: critical } return None def execute_all_rules(self, context): 执行所有规则返回最高优先级动作 actions [] for rule_id, rule_func in self.rules.items(): try: action rule_func(context) if action: actions.append(action) except Exception as e: print(f规则{rule_id}执行异常: {e}) if not actions: return None # 按priority排序critical high medium priority_map {critical: 3, high: 2, medium: 1} actions.sort(keylambda x: priority_map.get(x[priority], 0), reverseTrue) return actions[0] # 返回最高优先级动作 # 使用示例 # context { # lstm_prediction: [8500, 8700, 8900], # 15/30/60min预测 # xgb_shap: shap_result, # XGBoost的SHAP分析结果DataFrame # isoforest_result: df_labeled, # 孤立森林标注结果 # raw_data: df_today, # 当日原始数据 # load_threshold: 8000 # } # engine EnergyRuleEngine() # action engine.execute_all_rules(context) # if action: # print(f触发动作: {action[action]}, 目标设备: {action[target_devices]})规则引擎设计哲学每个规则函数独立便于单元测试和业务方审核如让节能工程师确认“空调贡献度60%才建议关机”是否合理context字典传递所有上游模型结果解耦模型与规则模型升级不影响规则逻辑关键技巧execute_all_rules返回最高优先级动作而非全部动作——避免GUI同时弹出10个窗口这是某次测试时运维抱怨“告警比微信消息还多”的直接改进。4.2 GUI设计用PyQt5构建生产级监控界面重点在“一键导出报告”和“设备图谱联动”GUI不是炫技核心功能就两个实时看板趋势异常点 一键生成PDF节能报告。我们放弃Web方案部署复杂、离线不可用用PyQt5构建桌面应用所有图表用Matplotlib嵌入设备拓扑用QGraphicsScene绘制。# gui/main_window.py import sys from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QTabWidget, QLabel, QPushButton, QComboBox, QGroupBox, QGridLayout, QFileDialog) from PyQt5.QtCore import Qt, QTimer import matplotlib.pyplot as plt from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas import pandas as pd from reportlab.pdfgen import canvas from reportlab.lib.pagesizes import A4 from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Image from reportlab.lib.styles import getSampleStyleSheet class EnergyMonitorGUI(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle(电力能耗智能分析系统 v1.0) self.setGeometry(100, 100, 1400, 800) # 主布局 central_widget QWidget() self.setCentralWidget(central_widget) main_layout QVBoxLayout(central_widget) # 顶部控制栏 control_layout QHBoxLayout() self.device_combo QComboBox() self.device_combo.addItems([METER_001, METER_002, TRANSFORMER_A]) self.date_picker QLabel(2024-07-23) self.refresh_btn QPushButton(刷新数据) self.refresh_btn.clicked.connect(self.load_data) control_layout.addWidget(QLabel(设备:)) control_layout.addWidget(self.device_combo) control_layout.addWidget(QLabel(日期:)) control_layout.addWidget(self.date_picker) control_layout.addWidget(self.refresh_btn) control_layout.addStretch() main_layout.addLayout(control_layout) # 选项卡 self.tabs QTabWidget() main_layout.addWidget(self.tabs) # 实时趋势页 self.trend_tab QWidget() self.trend_layout QVBoxLayout(self.trend_tab) self.trend_canvas MatplotlibCanvas() self.trend_layout.addWidget(self.trend_canvas) self.tabs.addTab(self.trend_tab, 实时趋势) # 异常分析页 self.anomaly_tab QWidget() self.anomaly_layout QVBoxLayout(self.anomaly_tab) self.anomaly_canvas MatplotlibCanvas() self.anomaly_layout.addWidget(self.anomaly_canvas) self.export_btn QPushButton(导出今日分析报告 (PDF)) self.export_btn.clicked.connect(self.export_report) self.anomaly_layout.addWidget(self.export_btn) self.tabs.addTab(self.anomaly_tab, 异常分析) # 加载初始数据 self.load_data() # 定时刷新每60秒 self.timer QTimer() self.timer.timeout.connect(self.load_data) self.timer.start(60000) def load_data(self): 从数据库加载最新数据并更新图表 device_id self.device_combo.currentText() # 伪代码实际从MySQL查询 # df_trend pd.read_sql(fSELECT * FROM energy_data WHERE device_id{device_id} AND timestamp NOW() - INTERVAL 24 HOUR, conn) # df_anomaly pd.read_sql(fSELECT * FROM energy_data WHERE device_id{device_id} AND is_anomaly1 AND date(timestamp)CURDATE(), conn) # 更新趋势图模拟数据 self.trend_canvas.plot_trend([i for i in range(288)], [1000500*i%100 for i in range(288)]) # 288个5分钟点 # 更新异常图模拟数据 self.anomaly_canvas.plot_anomaly([10, 50, 120, 200], [1200, 1500, 1800, 2100]) # 异常点位置和值 def export_report(self): 一键导出PDF报告 device_id self.device_combo.currentText() filename, _ QFileDialog.getSaveFileName( self, 保存报告, f{device_id}_energy_report_{self.date_picker.text()}.pdf, PDF Files (*.pdf) ) if not filename: return # 构建PDF内容 doc SimpleDocTemplate(filename, pagesizeA4) styles getSampleStyleSheet() story [] story.append(Paragraph(f设备 {device_id} 能耗分析报告, styles[Title])) story.append(Spacer(1, 12)) story.append(Paragraph(f日期: {self.date_picker.text()}, styles[Normal])) story.append(Spacer(1, 12)) # 添加趋势图需先保存为图片 trend_img_path temp_trend.png self.trend_canvas.figure.savefig(trend_img_path, dpi150, bbox_inchestight) story.append(Image(trend_img_path, width400, height250)) story.append(Spacer(1, 12)) # 添加关键结论从规则引擎获取 # action rule_engine.execute_all_rules(context) # 实际调用 # if action: # story.append(Paragraph(f【关键建议】{action[reason]}, styles[Heading2])) # story.append(Paragraph(f执行动作: {action[action]}, styles[Normal])) doc.build(story) print(f报告已导出: {filename}) class MatplotlibCanvas(FigureCanvas): def __init__( p a hrefhttps://download.csdn.net/download/xiaoxingkongyuxi/90437377 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p