智能电表 AMI 方案
AMI 概述
AMI（Advanced Metering Infrastructure，高级计量基础设施）是智能电网的重要组成部分，通过为电表加装通信模组，实现远程抄表、用电分析、需求响应等功能。
AMI vs 传统抄表
| 对比项 | 传统人工抄表 | AMI 智能抄表 |
|---|---|---|
| 抄表方式 | 人工上门 | 远程自动 |
| 抄表周期 | 每月一次 | 每15分钟一次 |
| 数据精度 | 人工误差 | 精确到 0.01kWh |
| 异常发现 | 滞后（下次抄表） | 实时告警 |
| 运营成本 | 高（人力） | 低（自动化） |
| 数据价值 | 低（月度数据） | 高（15分钟粒度） |
国内智能电表标准
国网标准
DL/T 645-2007：多功能电能表通信规约
- 物理层：RS485
- 波特率：1200/2400/4800/9600 bps
- 帧格式：68H + 地址 + 68H + 控制码 + 数据长度 + 数据 + CS + 16H
HPLC（高速电力线载波）：
- 国网主推通信方式
- 频段：2-12MHz
- 速率：最高 2Mbps
- 特点：利用电力线传输，无需额外布线
无线通信（补充方案）：
- NB-IoT：偏远地区、信号覆盖好的区域
- 4G Cat.1：数据量较大的场景
- LoRa：私有网络部署电表数据项
基本电量数据：
正向有功总电量（kWh）
反向有功总电量（kWh）
正向无功总电量（kvarh）
分时电量（峰/平/谷/尖）
实时数据：
电压（V）：A/B/C 三相
电流（A）：A/B/C 三相
有功功率（kW）
无功功率（kvar）
功率因数
频率（Hz）
事件记录：
失压事件
失流事件
过压/欠压事件
停电/来电事件
编程事件（防篡改）移远 NB-IoT 抄表方案
系统架构
┌─────────────────────────────────────────────────────┐
│ 智能电表（含通信模组） │
│ 电能计量芯片 + MCU + BC660K（NB-IoT） │
└─────────────────────────────────────────────────────┘
↓ NB-IoT 网络
┌─────────────────────────────────────────────────────┐
│ 运营商 NB-IoT 网络 │
└─────────────────────────────────────────────────────┘
↓ MQTT / CoAP
┌─────────────────────────────────────────────────────┐
│ 用电信息采集系统（主站） │
│ 国网 698 协议 / 自定义协议 │
└─────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────┐
│ 营销业务系统 │
│ 用电分析 / 账单生成 / 异常告警 / 需求响应 │
└─────────────────────────────────────────────────────┘推荐模组
BC660K-GL（NB-IoT，高通平台）：
优点：
- 超低功耗（PSM 2.5μA）
- 全球频段，覆盖广
- 高通平台稳定可靠
适用场景：
- 居民用电表（低频抄表，每天一次）
- 偏远地区（NB-IoT 覆盖增强）
BC26（NB-IoT，展锐平台）：
优点：
- 国产化，成本低
- 符合国网采购要求
适用场景：
- 大规模国网项目
- 成本敏感场景抄表数据上报
c
// NB-IoT 电表数据上报（QuecOpen）
#include "ql_api_common.h"
#include "ql_api_datacall.h"
#include "ql_api_coap.h"
#define METER_ID "320100123456" // 电表资产编号
#define COAP_SERVER "coap://iot.sgcc.com.cn:5683"
typedef struct {
char meter_id[20];
float total_energy; // 总有功电量（kWh）
float peak_energy; // 峰电量
float flat_energy; // 平电量
float valley_energy; // 谷电量
float voltage_a; // A相电压
float current_a; // A相电流
float power; // 有功功率
float power_factor; // 功率因数
uint32_t timestamp;
} MeterData;
// 读取电表数据（DL/T 645-2007 协议）
MeterData read_meter_data(void) {
MeterData data = {0};
strncpy(data.meter_id, METER_ID, sizeof(data.meter_id));
// 通过 RS485 读取电表
// 数据标识：00 00 00 00 = 正向有功总电量
uint8_t request[] = {
0x68, // 起始符
0x56, 0x34, 0x12, 0x00, 0x00, 0x00, // 电表地址（BCD）
0x68, // 起始符
0x11, // 控制码（读数据）
0x04, // 数据长度
0x33, 0x33, 0x33, 0x33, // 数据标识（总有功电量）
0x00, // CS（需计算）
0x16 // 结束符
};
// 计算校验和
request[13] = calculate_cs(request, 13);
// 发送请求
uart_send(request, sizeof(request));
// 接收响应并解析
uint8_t response[64];
int len = uart_recv(response, sizeof(response), 500);
if (len > 0) {
// 解析 BCD 编码的电量数据
data.total_energy = bcd_to_float(&response[10], 4) / 100.0f;
}
data.timestamp = get_unix_timestamp();
return data;
}
// 上报数据
void report_meter_data(void) {
MeterData data = read_meter_data();
// 构建 JSON 载荷
char payload[256];
snprintf(payload, sizeof(payload),
"{\"id\":\"%s\",\"e\":%.2f,\"ep\":%.2f,\"ef\":%.2f,"
"\"ev\":%.2f,\"v\":%.1f,\"i\":%.2f,\"p\":%.3f,\"pf\":%.3f,\"t\":%u}",
data.meter_id, data.total_energy, data.peak_energy,
data.flat_energy, data.valley_energy,
data.voltage_a, data.current_a, data.power,
data.power_factor, data.timestamp);
// CoAP POST 上报
Ql_coapPost(COAP_SERVER "/meter/data", payload, strlen(payload));
}
// 主程序：每天凌晨0点抄表上报
void ql_app_main(void) {
// 等待网络
wait_network_ready();
while (1) {
// 立即上报一次
report_meter_data();
// 计算到下一个整点的等待时间
uint32_t now = get_unix_timestamp();
uint32_t next_report = (now / 86400 + 1) * 86400; // 下一个0点
uint32_t sleep_seconds = next_report - now;
// 进入 PSM 睡眠
enter_psm(sleep_seconds);
}
}用电异常检测
窃电检测算法
python
import numpy as np
from sklearn.ensemble import IsolationForest
class StealingDetector:
"""基于机器学习的窃电检测"""
def __init__(self):
self.model = IsolationForest(
contamination=0.05, # 预期异常比例 5%
random_state=42
)
self.is_trained = False
def train(self, historical_data):
"""用历史正常数据训练模型"""
features = self._extract_features(historical_data)
self.model.fit(features)
self.is_trained = True
def _extract_features(self, data):
"""提取特征"""
features = []
for record in data:
feature = [
record['daily_energy'], # 日用电量
record['peak_ratio'], # 峰谷比
record['power_factor'], # 功率因数
record['voltage_unbalance'], # 三相不平衡度
record['current_unbalance'], # 电流不平衡度
record['reactive_ratio'], # 无功比例
]
features.append(feature)
return np.array(features)
def detect(self, meter_data):
"""检测是否异常"""
if not self.is_trained:
return False, 0.0
features = self._extract_features([meter_data])
score = self.model.score_samples(features)[0]
is_anomaly = self.model.predict(features)[0] == -1
return is_anomaly, abs(score)
def analyze_anomaly(self, meter_data):
"""分析异常原因"""
reasons = []
# 功率因数异常（可能绕过计量）
if meter_data['power_factor'] < 0.5:
reasons.append("功率因数异常偏低，疑似绕过计量装置")
# 三相不平衡（可能单相窃电）
if meter_data['voltage_unbalance'] > 5:
reasons.append("三相电压严重不平衡，疑似单相窃电")
# 用电量突降（可能私接旁路）
if meter_data['daily_energy'] < meter_data['avg_energy'] * 0.3:
reasons.append("用电量突降超70%，疑似私接旁路")
return reasons需求响应
削峰填谷
python
class DemandResponse:
"""需求响应：电网调度用电负荷"""
def handle_dr_signal(self, signal):
"""处理需求响应信号"""
if signal['type'] == 'peak_reduction':
# 削峰：通知用户减少用电
self.notify_users_reduce_load(
target_reduction_kw=signal['target_kw'],
duration_minutes=signal['duration'],
incentive_yuan_per_kwh=signal['incentive']
)
elif signal['type'] == 'valley_filling':
# 填谷：鼓励用户增加用电（如充电桩）
self.notify_users_increase_load(
target_increase_kw=signal['target_kw'],
duration_minutes=signal['duration'],
discount_percent=signal['discount']
)
def notify_users_reduce_load(self, target_reduction_kw,
duration_minutes, incentive_yuan_per_kwh):
"""通知用户减少负荷"""
# 通过 NB-IoT 下发控制命令到智能插座/空调控制器
command = {
"action": "reduce_load",
"target_reduction_percent": 20,
"duration_minutes": duration_minutes,
"incentive": incentive_yuan_per_kwh
}
# 广播给参与需求响应的用户
for user in self.get_dr_participants():
self.send_command_to_device(user['device_id'], command)