基于Python的电能质量监测系统设计与实现的详细项目实例

请注意此篇内容只是一个项目介绍 更多详细内容可直接联系博主本人 

 或者访问对应标题的完整博客或者文档下载页面（含完整的程序，GUI设计和代码详解）

电能质量监测系统面向现代电力环境中的电压波动、谐波超标、电压暂降、频率偏移、三相不平衡、闪变等问题而设计。随着工业自动化、数据中心、轨道交通、医院、精密制造和新能源接入规模持续扩大，电网负荷结构越来越复杂，非线性设备和冲击性负载显著增加，传统仅关注电能计量的方式已无法满足精细化运维需求。电能质量不仅影响设备寿命、生产连续性和用电安全，还直接关联停机损失、产品良率、能源浪费以及电网稳定性。对于配电房、变电站、工厂总进线、重点车间、医院手术室、通信机房和商业综合体而言，实时掌握电压、电流、功率因数、频率、谐波含量以及异常事件的发生时间、持续时长和严重程度，已经成为基础能力。

Python 作为数据采集、信号处理、特征分析、可视化展示和服务开发的统一技术栈，能够将采样数据处理、事件识别、统计报表、告警推送和网页端呈现整合到同一系统中。相较于传统分散式脚本与独立软件，基于 Python 的监测系统更适合快速构建原型、持续迭代算法以及低成本对接 Modbus、MQTT、OPC UA、串口与文件流。系统可通过采样模块持续接收来自电能质量分析仪、智能电表、互感器和边缘网关的数据，再经过预处理、指标计算、异常识别与数据库落库，最终在 Web 界面、移动端页面或大屏上显示趋势曲线、事件列表、告警状态与历史报表。该类系统在新型电力系统和智能配电数字化建设中，已经从“可选工具”逐步转变为“必备基础设施”。

在实际工程中，电能质量监测的难点并不只是数据采集，还包括采样精度、时间同步、瞬态事件识别、长时统计与多源数据融合。比如电压暂降可能仅持续几十毫秒，如果采样频率不足或时间戳不准确，事件就会被遗漏；再如谐波分析若缺少有效滤波和窗口处理，频谱结果会失真；三相不平衡若仅观察单点值，无法判断问题是持续性故障还是偶发负载冲击。因此，一个完善的系统不仅需要界面漂亮，更需要在算法、时序、存储、告警、权限和扩展性上形成闭环。基于 Python 构建的项目实例能够将理论算法与工程实现结合起来，完成从数据接入到分析展示的完整链路，为电力运维、能效管理和故障定位提供可落地的参考方案。

项目目标与意义

实时采集与统一接入

系统首要目标是将来自不同设备、不同协议、不同采样频率的电能质量数据统一接入到同一处理框架中。实际现场中，智能电表、谐波监测装置、微机保护装置、边缘采集终端以及网关设备往往来源不一，通信方式也各不相同。若缺少统一接入机制，后续分析会出现字段不一致、时间对不齐、数据粒度不同等问题。通过 Python 构建统一采集层，可以将串口、Modbus TCP、MQTT、HTTP API 与本地文件等多源输入封装成统一的数据对象，形成稳定的数据流通道。这样不仅能减少接入成本，还能为后续算法处理建立一致的数据标准，使得系统可以更轻松地扩展到多变电站、多楼宇、多产线的场景，保证电能质量监测具备工程可复制性和可维护性。

异常识别与告警联动

电能质量监测的核心价值不在于展示普通曲线，而在于尽早发现异常并快速触发处理流程。系统目标之一是建立基于阈值、滑动窗口、统计特征与事件规则的异常识别能力，对电压骤降、骤升、频率偏离、总谐波畸变率升高、三相电压不平衡、负载冲击等现象进行自动检测。识别结果不仅显示在界面中，还要能够联动消息通知、日志记录、事件归档和运维工单。这样一来，值班人员可以在异常发生后的第一时间接收到具体事件、发生时间、持续时长和严重等级，而不是等到设备损坏或产线停机后再被动排查。告警联动让系统从“监测工具”升级为“主动防护工具”，在降低事故概率、缩短排障时间、减少经济损失方面具有直接意义。

数据分析与运维决策

系统的另一个重要目标是将原始电参量转化为可用于决策的分析结果。长期运行的电能质量数据包含大量趋势信息，例如某些时段谐波持续偏高、某条支路电压偏差较大、某类设备投运时功率因数明显下降、某一周夜间暂降事件频发等。通过统计分析、趋势分析、分时段聚合、峰谷变化比较与报表生成，可以帮助运维人员识别问题规律并制定治理措施。例如加装滤波器、调整无功补偿、优化负荷调度、检查接地与接线、更新老旧设备等。这样做的意义不仅在于故障处理，更在于形成面向能效和可靠性的长期优化闭环，使电力管理从经验驱动转变为数据驱动，为节能降耗和设备健康管理提供依据。

教学研究与工程示范

基于 Python 的电能质量监测系统具有很强的教学与研究价值。对于高校课程设计、毕业设计、实验平台和科研验证而言，该系统能够把信号处理、数据库、Web 开发、工业通信和可视化多个知识点整合在一个真实场景中，让理论学习与工程实践同步进行。对于企业内部培训和技术验证，该系统可以作为边缘计算与工业数据处理的示范项目，帮助研发人员理解从采样到告警的完整链路。更重要的是，项目可通过模块化设计进行二次开发，例如加入机器学习分类、深度学习异常检测、数字孪生展示、移动端推送、地图分区监测等功能，具有很强的演进潜力。其意义不只是完成一个程序，而是搭建一套能够持续扩展、持续优化、持续落地的电能质量数字化底座。

项目挑战及解决方案

数据采样精度与时间同步

电能质量事件往往发生得非常短暂，尤其是电压暂降、暂升和冲击电流，持续时间可能远低于常规工频趋势数据的采样周期。如果采样间隔过大，关键事件就会被平均掉，导致系统误判为正常状态。解决这一问题的关键在于提高采样频率、合理设置窗口长度，并在数据接入层加入统一时间戳机制。Python 端可在接收原始帧时立即记录毫秒级时间，必要时结合 NTP 校时或网关同步方式校准设备时间。对于多源数据，还需要引入缓存队列和顺序重排，避免网络延迟导致的乱序。通过高频采样、同步校时、时间对齐和缓存重组，才能保证事件检测的准确性和统计结果的可信度。

谐波分析与异常判别

谐波分析的难点在于原始波形中可能包含噪声、非稳态变化和采样抖动，直接进行频域分析容易受到窗函数和边界效应影响。解决方案是先对信号进行去噪、标准化与滑动窗口切片，再采用快速傅里叶变换计算各次谐波幅值，进而计算总谐波畸变率。若希望提升鲁棒性，还可以对多个窗口结果进行均值或中位数聚合，避免单个异常点干扰整体判断。异常判别方面，系统应采用多条件联合判断，而不是仅靠单一阈值。例如当 THD 超过标准上限且持续多个周期时才生成告警；当电压偏差和频率偏差同时出现时，提升告警等级。这样的策略能有效减少误报与漏报，使监测结果更符合现场实际。

系统扩展性与可维护性

电能质量监测系统通常会随着设备增多、点位增多和功能要求提升而持续演化。如果前期结构设计过于耦合，后期新增协议支持、报表功能或算法模块就会非常困难。解决这一问题的核心是模块化与分层设计。数据采集、预处理、分析、存储、接口和展示应当分成独立层级，每一层只处理自身职责范围内的任务。Python 中可通过类封装、函数抽象、配置文件和统一日志管理来降低耦合度，并用数据库模型保存结构化结果，用 API 服务暴露标准化接口，用前端页面负责展示。这样即便后期替换数据库、增加新设备、改造算法，整体架构也能保持稳定，不会因局部变化导致全局重写。对于实际项目而言，这种可维护性决定了系统能否长期运行。

项目模型架构

数据采集层

数据采集层负责从电表、监测仪、模拟源或文件流中获取原始电参量数据，包含电压、电流、频率、有功功率、无功功率、视在功率、功率因数以及波形采样值。其核心任务是保证数据完整、格式统一、时间准确。工程上常见的做法是把不同来源的数据转换为统一字典结构，再进入处理管道。采集层可以使用串口通信、Socket、MQTT 客户端或文件监听方式实现，既可面对真实设备，也可对接仿真数据源。该层的关键算法原理在于“统一接口抽象”，即把多种输入转换为同一种输出格式，从而为后续算法提供稳定输入。采集层若设计合理，系统就能够轻松扩展更多点位和更多协议，且不会影响后续分析逻辑。

数据预处理层

预处理层主要完成缺失值处理、异常值剔除、时间对齐、平滑滤波与归一化。电能质量数据常因通信抖动、采集失败或设备异常而出现空值和突变值，若不先清洗，后续计算会严重失真。常用算法包括移动平均滤波、中值滤波、插值补全和阈值裁剪。移动平均适合平滑随机噪声，中值滤波适合去除尖峰脉冲，插值补全适合修复短时间缺失数据。时间对齐则是将不同采样频率的数据统一到同一时间轴上，便于统计分析和趋势绘图。预处理层的本质是提升数据质量，让原始数据变得可计算、可比较、可存储，从而确保电能质量指标具有工程意义。

特征计算层

特征计算层是系统的算法核心，负责把原始波形和电参量转换为可解释的质量指标。典型特征包括 RMS 有效值、峰值、频率偏差、功率因数、三相不平衡度、总谐波畸变率、单次谐波含量以及事件持续时间。RMS 反映交流量等效能量水平，功率因数描述有功与视在功率之间的利用效率，THD 则衡量波形偏离正弦的程度。频域分析一般依赖快速傅里叶变换，利用其将时域信号映射到频域，从而提取基波与谐波分量。三相不平衡计算通常依据三相电压或电流的偏差比例，衡量系统是否存在严重偏相。通过这些特征，系统能够将大量原始采样压缩成少量高价值指标，便于存储、查询与告警。

事件识别层

事件识别层基于特征指标和规则引擎判断是否存在异常电能质量事件。它可以采用阈值法、滑动窗口法、连续超限判定法、变化率判定法等策略。阈值法实现简单，适合标准明确的监测指标，例如电压越限、频率越界、THD 超限；滑动窗口法适用于识别持续性波动和短周期异常；连续超限法可减少瞬时抖动造成的误报；变化率判定法则可捕捉骤降、骤升和冲击类事件。该层的原理是把单次测量值提升为时序行为判断，强调事件的持续性、重复性和严重性。通过多规则融合，系统能够更准确地区分正常波动与真正故障，提高告警质量。

数据存储与展示层

数据存储与展示层承担历史追溯、统计查询和人机交互功能。结构化数据可存入 SQLite、MySQL 或 PostgreSQL，原始波形和大容量采样数据可采用文件系统或时序数据库保存。展示层则通过 Flask、FastAPI 或 Django 提供接口，再配合前端图表组件绘制趋势曲线、频谱图、告警表和统计面板。存储层的关键原理是按时间与设备维度建模，保证查询高效；展示层的关键原理是将复杂数据可视化成直观图表，使运维人员快速理解系统状态。一个完整架构必须同时兼顾实时性、可靠性与可视化效果，这样才能从工程工具升级为真正可用的监测平台。

项目模型描述及代码示例

数据采集模型
电能质量监测的起点是稳定获取数据。该模型通过模拟源生成电压、电流与频率等参数，便于在无真实设备环境中验证整套流程。采集对象应统一成字典结构，字段中包含时间戳、设备编号、三相电压、三相电流、频率、功率因数和波形数组。这样做的好处是后续处理可以直接按照字段名读取，不会因为来源不同而改动逻辑。采集层的程序重点是可扩展、易替换，既能对接真实设备，也能在调试阶段使用模拟数据稳定运行。每次采集完成后立即打印或写入日志，便于确认数据流是否正常。对于实际部署，采集层还可增加异常重试、断线重连与设备心跳检测。
import time # 导入时间模块，用于生成采样间隔和时间戳
import math # 导入数学模块，用于生成正弦波和周期信号
import random # 导入随机模块，用于模拟噪声和扰动
from datetime import datetime # 导入日期时间模块，用于记录采样时刻
def generate_sample(device_id="PQ-001", t=0): # 定义单条采样数据生成函数，device_id 表示设备编号，t 表示当前采样序号
base_voltage = 220.0 # 设置基准电压为 220V，模拟常见低压系统的额定值
base_current = 10.0 # 设置基准电流为 10A，作为负载参考值
freq = 50.0 + random.uniform(-0.08, 0.08) # 生成轻微波动的频率，模拟电网实际偏移
voltage_a = base_voltage + 5 * math.sin(2 * math.pi * t / 100) + random.uniform(-1.5, 1.5) # 生成 A 相电压，包含周期波动和随机噪声
voltage_b = base_voltage - 3 + 4 * math.sin(2 * math.pi * t / 97) + random.uniform(-1.5, 1.5) # 生成 B 相电压，加入相间偏差
voltage_c = base_voltage + 2 + 6 * math.sin(2 * math.pi * t / 103) + random.uniform(-1.5, 1.5) # 生成 C 相电压，模拟不同负荷影响
current_a = base_current + 1.5 * math.sin(2 * math.pi * t / 30) + random.uniform(-0.3, 0.3) # 生成 A 相电流，包含负载波动
current_b = base_current * 0.95 + 1.3 * math.sin(2 * math.pi * t / 28) + random.uniform(-0.3, 0.3) # 生成 B 相电流，模拟轻微不平衡
current_c = base_current * 1.05 + 1.7 * math.sin(2 * math.pi * t / 32) + random.uniform(-0.3, 0.3) # 生成 C 相电流，模拟轻微不平衡
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # 生成当前时间字符串，作为记录时间
waveform = [math.sin(2 * math.pi * 50 * x / 2000) + 0.08 * math.sin(2 * math.pi * 150 * x / 2000) for x in range(200)] # 生成含三次谐波的波形，用于后续频谱分析
return { # 返回统一结构的数据字典，便于后续处理
"device_id": device_id, # 设备编号字段，用于区分采集源
"timestamp": timestamp, # 时间戳字段，用于时序分析
"voltage_a": round(voltage_a, 2), # A 相电压保留两位小数
"voltage_b": round(voltage_b, 2), # B 相电压保留两位小数
"voltage_c": round(voltage_c, 2), # C 相电压保留两位小数
"current_a": round(current_a, 2), # A 相电流保留两位小数
"current_b": round(current_b, 2), # B 相电流保留两位小数
"current_c": round(current_c, 2), # C 相电流保留两位小数
"frequency": round(freq, 3), # 频率保留三位小数，提升监测精度
"waveform": waveform # 保存波形样本，供频域分析使用
} # 结束字典返回
def acquire_stream(count=5): # 定义连续采集函数，count 表示采样条数
samples = [] # 创建列表保存采样结果
for i in range(count): # 循环生成指定数量的样本
samples.append(generate_sample(t=i)) # 调用生成函数并加入列表
time.sleep(0.1) # 模拟采样间隔，保持数据流节奏
return samples # 返回全部采样数据
data_batch = acquire_stream(3) # 调用采集函数生成三条样本
for item in data_batch: # 遍历样本列表
print(item) # 打印每条采样结果，确认数据结构正常
数据预处理模型
预处理模型的任务是让原始数据变得可计算。实际监测中可能出现局部缺失、偶发尖峰和不规则抖动，因此需先完成清洗再进入分析阶段。这里采用中值平滑处理波动异常，并对缺失值进行简单插值补全。对于电能质量场景，过度平滑会掩盖短时异常，所以平滑策略应尽量保守，只处理明显噪声而保留事件边缘。预处理完成后，数据更适合做趋势分析、谐波分析和超限判定。对于不同字段，可设置不同的处理规则，例如电压与频率适合平滑，事件标志与告警等级则不应改写。该模型能够作为后续算法的稳定入口。
import statistics # 导入统计模块，用于中值与均值计算
def clean_value_list(values): # 定义清洗函数，输入为数值列表
cleaned = [] # 创建清洗后的结果列表
for i, v in enumerate(values): # 遍历原始数据及索引
if v is None: # 判断是否存在缺失值
left = cleaned[-1] if cleaned else 0 # 获取左侧最近值，若无则使用 0
right = values[i + 1] if i + 1 < len(values) and values[i + 1] is not None else left # 获取右侧值，用于插值
cleaned.append((left + right) / 2) # 使用简单均值补全缺失值
else: # 如果不是缺失值
cleaned.append(v) # 直接保留原值
return cleaned # 返回补全后的列表
def median_smooth(values, window=3): # 定义中值平滑函数，window 表示窗口长度
if window < 3: # 判断窗口是否合理
return values[:] # 若窗口过小则直接返回原列表副本
half = window // 2 # 计算半窗口大小
smoothed = [] # 创建平滑后的结果列表
for i in range(len(values)): # 遍历每个位置
start = max(0, i - half) # 计算窗口起点
end = min(len(values), i + half + 1) # 计算窗口终点
smoothed.append(statistics.median(values[start:end])) # 计算局部中值并加入结果
return smoothed # 返回平滑列表
raw_series = [220.1, 219.8, None, 260.0, 220.2, 219.9, 220.0] # 构造一组包含缺失值和尖峰的数据
filled_series = clean_value_list(raw_series) # 对缺失值进行补全
filtered_series = median_smooth(filled_series, window=3) # 对数据进行中值平滑
print("原始序列:", raw_series) # 输出原始序列
print("补全序列:", filled_series) # 输出补全结果
print("平滑序列:", filtered_series) # 输出平滑结果
谐波分析模型
谐波分析用于衡量信号是否偏离理想正弦波。系统通过快速傅里叶变换将时域波形转换到频域，并从频谱中提取基波与高次谐波幅值，再计算总谐波畸变率。THD 越高，说明波形越不纯净，可能存在整流设备、变频器或开关电源等非线性负载。代码中先对波形做幅值归一化，再使用 numpy 的 FFT 计算频谱，最后通过谐波能量与基波能量比例得到 THD。实际系统可进一步提取 2、3、5、7 次谐波分别统计，辅助定位污染源。该方法适合对稳态波形进行批量分析，是电能质量系统最常用的指标计算方式之一。
import numpy as np # 导入 numpy，用于数组计算和傅里叶变换
def calculate_thd(signal): # 定义 THD 计算函数，输入为波形信号
arr = np.array(signal, dtype=float) # 将输入转换为浮点数组
arr = arr - np.mean(arr) # 去除直流分量，避免影响频谱分析
spectrum = np.fft.rfft(arr) # 对实数信号做快速傅里叶变换
amplitudes = np.abs(spectrum) # 计算频谱幅值
if len(amplitudes) < 3: # 判断频谱长度是否足够
return 0.0 # 若长度不足则返回 0
fundamental = amplitudes[1] # 取第一个谐波分量作为基波幅值
harmonic_power = np.sum(np.square(amplitudes[2:])) # 累加高次谐波能量
if fundamental == 0: # 判断基波是否为 0
return 0.0 # 避免除零错误
thd = np.sqrt(harmonic_power) / fundamental * 100 # 计算 THD 百分比
return round(float(thd), 2) # 返回保留两位小数的结果
wave = [math.sin(2 * math.pi * x / 32) + 0.12 * math.sin(2 * math.pi * 3 * x / 32) for x in range(256)] # 构造带三次谐波的示例波形
thd_value = calculate_thd(wave) # 计算谐波畸变率
print("THD:", thd_value) # 输出 THD 结果
电能质量指标计算模型
该模型用于计算有效值、功率因数和三相不平衡度等关键指标。RMS 反映交流量的实际能量水平，功率因数表示电能利用效率，三相不平衡度则衡量三相系统偏差情况。RMS 的计算基于平方平均再开方，适合处理波动信号；功率因数通常由有功功率与视在功率之比得到；不平衡度则通过最大偏差相对于平均值的比例体现。系统中这些指标通常定时计算，并与阈值比较后形成趋势与告警。该模型是将物理意义转化为业务判断的关键桥梁，也是运维人员最关注的核心结果。
import math # 再次导入数学模块，用于平方与开方计算
def rms(values): # 定义有效值计算函数
arr = np.array(values, dtype=float) # 将输入转为数组
return round(float(np.sqrt(np.mean(np.square(arr)))), 3) # 按 RMS 公式计算并保留三位小数
def power_factor(active_power, apparent_power): # 定义功率因数计算函数
if apparent_power == 0: # 判断视在功率是否为 0
return 0.0 # 避免除零
return round(active_power / apparent_power, 3) # 计算功率因数并保留三位小数
def unbalance_degree(a, b, c): # 定义三相不平衡度计算函数
values = np.array([a, b, c], dtype=float) # 构造三相数组
avg = np.mean(values) # 计算三相平均值
if avg == 0: # 判断平均值是否为 0
return 0.0 # 避免除零
deviation = np.max(np.abs(values - avg)) # 计算最大偏差
return round(float(deviation / avg * 100), 2) # 返回不平衡百分比
ia = [10.1, 10.2, 10.3, 10.0, 9.9] # 构造电流样本
ua = [220.3, 219.9, 220.1, 220.0, 220.2] # 构造电压样本
print("电流RMS:", rms(ia)) # 输出电流有效值
print("电压RMS:", rms(ua)) # 输出电压有效值
print("功率因数:", power_factor(8500, 9200)) # 输出功率因数
print("三相不平衡度:", unbalance_degree(220.1, 218.0, 221.5)) # 输出三相不平衡度
异常检测模型
异常检测模型通过阈值规则和连续超限判断识别事件。系统中常见做法是对某个指标设置上限与下限，并在多个连续采样点都超限时才认定为异常，以降低瞬时抖动导致的误报。该方法虽然简单，但在工程场景中非常实用，尤其适合电压越限、频率偏移和 THD 超标等明确标准场景。代码中将连续三次超限视为异常事件，并返回触发位置与等级。后续可把事件写入数据库或触发消息推送。该模型是告警系统的基础，能帮助现场迅速定位风险区间，减少停机与损坏风险。
def detect_threshold_alarm(values, low=None, high=None, consecutive=3): # 定义阈值异常检测函数
counter = 0 # 初始化连续超限计数器
for idx, value in enumerate(values): # 遍历数据及索引
over_low = low is not None and value < low # 判断是否低于下限
over_high = high is not None and value > high # 判断是否高于上限
if over_low or over_high: # 若发生超限
counter += 1 # 连续计数加一
if counter >= consecutive: # 若连续超限达到设定次数
return { # 返回告警信息
"alarm": True, # 告警状态为真
"index": idx, # 记录触发位置
"value": value, # 记录触发值
"reason": "阈值连续超限" # 记录原因
} # 结束返回字典
else: # 若当前值正常
counter = 0 # 重置连续计数器
return { # 未触发告警时返回结果
"alarm": False, # 告警状态为假
"index": -1, # 未触发位置设为 -1
"value": None, # 未触发值设为 None
"reason": "正常" # 状态说明为正常
} # 结束返回字典
voltage_series = [219.8, 220.1, 220.4, 230.6, 231.2, 232.0, 220.0] # 构造电压序列
alarm_info = detect_threshold_alarm(voltage_series, low=210, high=230, consecutive=2) # 执行连续超限检测
print("告警结果:", alarm_info) # 输出告警信息
数据存储与服务模型
数据存储与服务模型负责把分析结果保留下来，并以接口形式提供给页面调用。这里使用 SQLite 作为演示数据库，原因是部署简单、无需额外安装服务、适合本地项目验证。程序先创建结果表，再把每条采样记录写入数据库。查询时可以按设备编号和时间范围筛选。服务层可通过 Flask 提供 JSON 接口，前端页面或大屏均可直接请求。该结构便于实现历史查询、统计报表与实时刷新。实际项目中也可替换为 MySQL 或 PostgreSQL，只需改动数据库连接部分即可，核心业务逻辑不变。
import sqlite3 # 导入 SQLite 模块，用于本地数据库存储
from flask import Flask, jsonify # 导入 Flask 框架与 JSON 响应工具
app = Flask(name) # 创建 Flask 应用实例
DB_PATH = "pq_monitor.db" # 设置数据库文件路径
def init_db(): # 定义数据库初始化函数
conn = sqlite3.connect(DB_PATH) # 连接数据库
cursor = conn.cursor() # 创建游标对象
cursor.execute(""" # 执行建表语句
CREATE TABLE IF NOT EXISTS pq_records ( # 创建电能质量记录表
id INTEGER PRIMARY KEY AUTOINCREMENT, # 主键自增
device_id TEXT, # 设备编号
timestamp TEXT, # 时间戳
voltage_a REAL, # A 相电压
voltage_b REAL, # B 相电压
voltage_c REAL, # C 相电压
current_a REAL, # A 相电流
current_b REAL, # B 相电流
current_c REAL, # C 相电流
frequency REAL, # 频率
thd REAL, # 谐波畸变率
alarm TEXT # 告警信息
) # 表结构结束
""") # 结束建表语句
conn.commit() # 提交数据库操作
conn.close() # 关闭连接
def save_record(record, thd, alarm): # 定义数据入库函数
conn = sqlite3.connect(DB_PATH) # 连接数据库
cursor = conn.cursor() # 创建游标
cursor.execute(""" # 执行插入语句
INSERT INTO pq_records (device_id, timestamp, voltage_a, voltage_b, voltage_c, current_a, current_b, current_c, frequency, thd, alarm)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) # 使用参数化防止注入
""", ( # 传入插入参数
record["device_id"], # 设备编号
record["timestamp"], # 时间戳
record["voltage_a"], # A 相电压
record["voltage_b"], # B 相电压
record["voltage_c"], # C 相电压
record["current_a"], # A 相电流
record["current_b"], # B 相电流
record["current_c"], # C 相电流
record["frequency"], # 频率
thd, # THD 值
alarm # 告警文本
)) # 完成插入参数
conn.commit() # 提交事务
conn.close() # 关闭连接
@app.route("/records") # 定义查询接口路由
def get_records(): # 定义接口函数
conn = sqlite3.connect(DB_PATH) # 连接数据库
cursor = conn.cursor() # 创建游标
cursor.execute("SELECT device_id, timestamp, voltage_a, voltage_b, voltage_c, frequency, thd, alarm FROM pq_records ORDER BY id DESC LIMIT 20") # 查询最新 20 条记录
rows = cursor.fetchall() # 获取所有查询结果
conn.close() # 关闭数据库连接
result = [ # 组织返回数据
{ # 每条记录转换为字典
"device_id": row[0], # 设备编号
"timestamp": row[1], # 时间戳
"voltage_a": row[2], # A 相电压
"voltage_b": row[3], # B 相电压
"voltage_c": row[4], # C 相电压
"frequency": row[5], # 频率
"thd": row[6], # THD
"alarm": row[7] # 告警信息
} for row in rows # 遍历查询结果
] # 结束列表生成
return jsonify(result) # 以 JSON 格式返回数据
init_db() # 初始化数据库
sample = generate_sample() # 生成一条样本
thd_val = calculate_thd(sample["waveform"]) # 计算样本 THD
alarm_text = detect_threshold_alarm([sample["voltage_a"], sample["voltage_b"], sample["voltage_c"]], low=210, high=235, consecutive=2)["reason"] # 生成告警文本
save_record(sample, thd_val, alarm_text) # 保存记录到数据库
print("数据库写入完成") # 输出写入结果
if name == "main": # 判断是否作为主程序运行
app.run(host="0.0.0.0", port=5000, debug=True) # 启动 Flask 服务，监听 5000 端口

更多详细内容请访问

http://基于Python的电能质量监测系统设计与实现的详细项目实例（含完整的程序，数据库和GUI设计，代码详解）【电力系统】基于Python的电能质量监测系统设计：_MATLAB时间序列预测代码资源-CSDN下载 https://download.csdn.net/download/xiaoxingkongyuxi/90242216

https://download.csdn.net/download/xiaoxingkongyuxi/90242216

https://download.csdn.net/download/xiaoxingkongyuxi/90242216