怎样对接智能大功率断路器计量数显版40A来实现实时电量计量统计_解决方案

CATALOG

[91619] 文档版本：V1.0 更新日期：2026年5月 适用对象：软件工程师、能源管理系统集成商、智能配电项目负责人

第一章：场景痛点与解决概述

1.1 电量计量统计的业务需求

在智能配电、共享充电、工业能耗管理等场景中，精准的电量计量统计是实现精细化运营和成本核算的基础。传统方案存在以下痛点：

痛点	具体表现	带来的问题
计量缺失	传统断路器仅支持通断控制，无计量功能	无法统计设备/线路的实际用电量
数据孤岛	电表数据与控制系统分离	无法实现基于电量的联动控制和告警
实时性差	人工抄表周期长	无法及时发现用电异常和浪费
改造成本高	加装独立电表需额外布线和设备	项目周期长，投资回报低

1.2 解决方案：计量数显版断路器实现电量统计

芯步智能大功率断路器[计量数显版]40A集成了电能计量模块，通过HTTP接口开放计量数据，实现“实时电量采集-数据存储分析-异常预警联动”的完整计量统计闭环。

核心价值

一体化设计：通断控制 + 电能计量二合一，无需额外电表
实时数据：可随时查询当前功率、累计用电量
远程可读：通过HTTP接口获取计量数据，支持任何编程语言
私有化部署：支持纯局域网运行，数据不出园区

1.3 智能大功率断路器[计量数显版]40A核心参数

本方案产品详细参数如下

参数项	规格详情
产品型号	UNI-DLQ-M-40A-PD
额定电流	MAX 40A
额定功率	MAX 8000W（阻性负载）/ 1400W（感性负载）
计量功能	功率计量版，实时上报计量数据
工作电压	AC 85-265V
安装方式	35mm DIN导轨式
联网方式	WiFi 2.4GHz
待机功耗	0.8W
外壳材质	防火V0级PC，耐高温，符合安规标准

版本说明：该产品提供WiFi版、WiFi+433遥控版、4G版等多个版本，可根据网络环境选择。

第二章：HTTP接口协议详解

2.1 整体架构

芯步智能断路器采用标准的HTTP请求-响应模型。设备开放HTTP接口，适用于任何支持HTTP请求的编程语言，支持接入Web、APP/小程序、窗体软件、SaaS/低代码等任何形式的软件项目。

graph LR
    subgraph 软件层
        A[能源管理系统/后台]
    end
    
    subgraph 网络层
        B[芯步API网关
api.thingboot.com]
    end
    
    subgraph 硬件层
        C[智能大功率断路器40A
计量数显版]
    end
    
    A -- "HTTP POST (带签名)" --> B
    B -- "MQTT透传" --> C
    C -- "返回计量数据" --> A

2.2 请求地址格式

http(s)://api.thingboot.com/{AppId}/device/control/?sign={sign}&ts={ts}

URL参数说明

参数	说明	示例
`{AppId}`	应用ID，在芯步控制台获取	`10001`
`sign`	动态签名，验证请求合法性	`a1b2c3d4...`
`ts`	当前Unix时间戳（秒级）	`1746000000`

2.3 签名算法（核心安全机制）

芯步API采用双重MD5签名机制，确保接口调用的安全性

Sign = md5( md5(AppSecret) + ts )

签名计算步骤

flowchart LR
    A[AppSecret] --> B[MD5加密]
    B --> C[32位小写: md5_secret]
    C --> D[拼接ts: md5_secret + ts]
    D --> E[再次MD5加密]
    E --> F[最终Sign]

核心代码实现

安全要点

两次MD5输出均为32位小写十六进制字符串
ts需与服务端时间同步，误差不超过5分钟
AppSecret需妥善保管，切勿硬编码在前端代码中

2.4 核心命令集

智能大功率断路器[计量数显版]支持以下核心命令

命令类别	命令格式	说明
通断控制	`{"power":"1"}`	接通/断开电源（1接通，0断开）
计量查询	`{"metering":""}`	查询电量计量数据
状态查询	`{"get_status":""}`	查询设备在线及通断状态
先通后断	`{"point":"30000"}`	通电30秒后自动断电
先断后通	`{"reset":"5000"}`	断电5秒后自动通电

2.5 计量数据查询（核心功能）

对于电量计量统计场景，最核心的功能是查询计量数据

命令格式

返回示例（根据实际产品定义）：

注意：实际返回字段请以产品手册为准，不同批次产品的计量数据格式可能略有差异。

2.6 完整请求示例

响应时间：从命令下发到设备响应约为80-120ms，响应非常快。

第三章：多语言代码实现

3.1 Python完整实现

# -*- coding: utf-8 -*-
import hashlib
import time
import requests
import json
from typing import Dict, Optional, List
from datetime import datetime
import sqlite3
import schedule

# ========== 配置信息（从芯步控制台获取） ==========
APP_ID = "YOUR_APP_ID"           # 应用ID
APP_SECRET = "YOUR_APP_SECRET"   # 应用密钥
DEVICE_ID = "YOUR_DEVICE_ID"     # 断路器设备ID
API_BASE = "https://api.thingboot.com"
# ===================================================

def calculate_md5(s: str) -> str:
    """计算MD5（32位小写）"""
    return hashlib.md5(s.encode('utf-8')).hexdigest()

def send_command(order: dict) -> dict:
    """向设备发送命令"""
    ts = int(time.time())
    md5_secret = calculate_md5(APP_SECRET)
    sign_raw = md5_secret + str(ts)
    sign = calculate_md5(sign_raw)
    
    url = f"{API_BASE}/{APP_ID}/device/control/"
    params = {"sign": sign, "ts": ts}
    payload = {"device": DEVICE_ID, "order": order}
    
    try:
        response = requests.post(url, params=params, json=payload, timeout=5)
        return response.json() if response.ok else {"error": response.text}
    except Exception as e:
        return {"error": str(e)}

# ========== 计量数据查询（核心功能）==========
def get_metering_data() -> Dict:
    """
    获取电量计量数据
    
    返回:
        {
            "power": float,     # 当前功率(W)
            "energy": float,    # 累计用电量(kWh)
            "voltage": float,   # 当前电压(V)
            "current": float,   # 当前电流(A)
            "online": bool      # 设备在线状态
        }
    """
    result = send_command({"metering": ""})
    
    if result.get("error"):
        return {"error": result["error"], "online": False}
    
    data = result.get("data", {})
    return {
        "power": float(data.get("power", 0)),
        "energy": float(data.get("energy", 0)),
        "voltage": float(data.get("voltage", 0)),
        "current": float(data.get("current", 0)),
        "online": data.get("online") == "1",
        "raw_data": data
    }

def get_device_status() -> Dict:
    """获取设备通断状态"""
    result = send_command({"get_status": ""})
    if result.get("error"):
        return {"error": result["error"]}
    return result.get("data", {})

def set_power(on: bool) -> Dict:
    """控制通断"""
    value = "1" if on else "0"
    return send_command({"power": value})

# ========== 电量统计与数据持久化 ==========
class EnergyMeter:
    """电量计量统计管理器"""
    
    def __init__(self, db_path: str = "energy_data.db"):
        self.db_path = db_path
        self._init_database()
    
    def _init_database(self):
        """初始化数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 创建计量数据表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS metering_data (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp DATETIME NOT NULL,
                power REAL,
                energy REAL,
                voltage REAL,
                current REAL,
                device_id TEXT
            )
        ''')
        
        # 创建日统计表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS daily_stats (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                stat_date DATE NOT NULL,
                total_energy REAL DEFAULT 0,
                max_power REAL DEFAULT 0,
                avg_power REAL DEFAULT 0,
                record_count INTEGER DEFAULT 0,
                device_id TEXT,
                UNIQUE(stat_date, device_id)
            )
        ''')
        
        conn.commit()
        conn.close()
    
    def save_metering_data(self, data: Dict, device_id: str = None):
        """保存计量数据到数据库"""
        if data.get("error") or not data.get("online"):
            return
        
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO metering_data (timestamp, power, energy, voltage, current, device_id)
            VALUES (?, ?, ?, ?, ?, ?)
        ''', (
            datetime.now(),
            data.get("power"),
            data.get("energy"),
            data.get("voltage"),
            data.get("current"),
            device_id or DEVICE_ID
        ))
        
        conn.commit()
        conn.close()
    
    def get_energy_consumption(self, start_date: str, end_date: str) -> float:
        """获取指定时间段的用电量（kWh）"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 获取时间段首尾的累计用电量差值
        cursor.execute('''
            SELECT energy FROM metering_data 
            WHERE timestamp >= ? AND timestamp <= ?
            ORDER BY timestamp 
            LIMIT 1
        ''', (start_date, end_date))
        start_energy = cursor.fetchone()
        
        cursor.execute('''
            SELECT energy FROM metering_data 
            WHERE timestamp >= ? AND timestamp <= ?
            ORDER BY timestamp DESC
            LIMIT 1
        ''', (start_date, end_date))
        end_energy = cursor.fetchone()
        
        conn.close()
        
        if start_energy and end_energy:
            return end_energy[0] - start_energy[0]
        return 0
    
    def update_daily_stats(self):
        """更新日统计数据"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        today = datetime.now().date()
        
        # 获取今日所有记录
        cursor.execute('''
            SELECT power, energy FROM metering_data 
            WHERE DATE(timestamp) = ?
        ''', (today,))
        records = cursor.fetchall()
        
        if records:
            total_energy = records[-1][1] - records[0][1] if len(records) > 1 else 0
            max_power = max(r[0] for r in records)
            avg_power = sum(r[0] for r in records) / len(records)
            
            cursor.execute('''
                INSERT OR REPLACE INTO daily_stats 
                (stat_date, total_energy, max_power, avg_power, record_count, device_id)
                VALUES (?, ?, ?, ?, ?, ?)
            ''', (today, total_energy, max_power, avg_power, len(records), DEVICE_ID))
        
        conn.commit()
        conn.close()

# ========== 定时采集与监控 ==========
class EnergyMonitor:
    """电量监控服务"""
    
    def __init__(self):
        self.meter = EnergyMeter()
        self.alert_threshold_power = 7000  # 功率告警阈值(W)
        self.alert_threshold_voltage_min = 190  # 欠压告警(V)
        self.alert_threshold_voltage_max = 240  # 过压告警(V)
    
    def collect_and_save(self):
        """采集并保存计量数据"""
        data = get_metering_data()
        
        if data.get("error"):
            print(f"[错误] 采集失败: {data['error']}")
            return
        
        self.meter.save_metering_data(data)
        print(f"[采集] {datetime.now()} - 功率:{data.get('power')}W, 累计:{data.get('energy')}kWh")
        
        # 检查告警条件
        self.check_alerts(data)
    
    def check_alerts(self, data: Dict):
        """检查告警条件"""
        alerts = []
        
        # 过载告警
        if data.get("power", 0) > self.alert_threshold_power:
            alerts.append(f"功率过高: {data['power']}W > {self.alert_threshold_power}W")
        
        # 电压异常告警
        voltage = data.get("voltage", 220)
        if voltage < self.alert_threshold_voltage_min:
            alerts.append(f"电压过低: {voltage}V < {self.alert_threshold_voltage_min}V")
        elif voltage > self.alert_threshold_voltage_max:
            alerts.append(f"电压过高: {voltage}V > {self.alert_threshold_voltage_max}V")
        
        for alert in alerts:
            self.send_alert(alert)
    
    def send_alert(self, message: str):
        """发送告警"""
        print(f"[告警] {datetime.now()}: {message}")
        # 可集成钉钉、企业微信、短信等通知渠道
    
    def start_collection(self, interval_seconds: int = 60):
        """启动定时采集"""
        def collect_job():
            self.collect_and_save()
        
        schedule.every(interval_seconds).seconds.do(collect_job)
        print(f"[服务] 电量采集服务已启动，采集间隔{interval_seconds}秒")
        
        while True:
            schedule.run_pending()
            time.sleep(1)

# ========== 业务场景函数 ==========
def get_current_power() -> float:
    """获取当前功率"""
    data = get_metering_data()
    return data.get("power", 0)

def get_total_energy() -> float:
    """获取累计用电量"""
    data = get_metering_data()
    return data.get("energy", 0)

def get_daily_energy() -> float:
    """获取今日用电量"""
    meter = EnergyMeter()
    today = datetime.now().date()
    tomorrow = today + timedelta(days=1)
    return meter.get_energy_consumption(str(today), str(tomorrow))

# ========== 调用示例 ==========
if __name__ == "__main__":
    # 示例1:单次查询计量数据
    # data = get_metering_data()
    # print(f"当前功率: {data.get('power')}W")
    # print(f"累计用电: {data.get('energy')}kWh")
    # print(f"当前电压: {data.get('voltage')}V")
    # print(f"当前电流: {data.get('current')}A")
    
    # 示例2:启动定时采集服务
    # monitor = EnergyMonitor()
    # monitor.start_collection(interval_seconds=60)
    
    # 示例3:通断控制
    # set_power(True)   # 接通电源
    # set_power(False)  # 断开电源
    pass

3.2 Node.js实现

const crypto = require('crypto');
const axios = require('axios');
const sqlite3 = require('sqlite3').verbose();

// 配置
const config = {
    appId: 'YOUR_APP_ID',
    appSecret: 'YOUR_APP_SECRET',
    deviceId: 'YOUR_DEVICE_ID',
    apiUrl: 'https://api.thingboot.com'
};

// 生成签名
function generateSign(secret, ts) {
    const firstMd5 = crypto.createHash('md5').update(secret).digest('hex');
    const signRaw = firstMd5 + ts;
    return crypto.createHash('md5').update(signRaw).digest('hex');
}

// 发送命令
async function sendCommand(order) {
    const ts = Math.floor(Date.now() / 1000);
    const sign = generateSign(config.appSecret, ts);
    
    const url = `${config.apiUrl}/${config.appId}/device/control/`;
    const payload = { device: config.deviceId, order };
    
    try {
        const response = await axios.post(url, payload, {
            params: { sign, ts },
            timeout: 5000
        });
        return response.data;
    } catch (error) {
        return { error: error.message };
    }
}

// 获取计量数据
async function getMeteringData() {
    const result = await sendCommand({ metering: "" });
    
    if (result.error) {
        return { error: result.error, online: false };
    }
    
    const data = result.data || {};
    return {
        power: parseFloat(data.power || 0),
        energy: parseFloat(data.energy || 0),
        voltage: parseFloat(data.voltage || 0),
        current: parseFloat(data.current || 0),
        online: data.online === '1'
    };
}

// 控制通断
async function setPower(on) {
    return sendCommand({ power: on ? '1' : '0' });
}

// 定时采集
function startCollection(intervalSeconds = 60) {
    setInterval(async () => {
        const data = await getMeteringData();
        console.log(`[${new Date().toISOString()}] 功率: ${data.power}W, 累计: ${data.energy}kWh`);
        
        // 可在此保存到数据库
    }, intervalSeconds * 1000);
}

// 使用示例
// getMeteringData().then(console.log);
// startCollection(60);

3.3 Java实现

import java.security.MessageDigest;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.URI;
import com.google.gson.Gson;

public class EnergyMeterController {
    private static final String APP_ID = "YOUR_APP_ID";
    private static final String APP_SECRET = "YOUR_APP_SECRET";
    private static final String DEVICE_ID = "YOUR_DEVICE_ID";
    private static final String API_BASE = "https://api.thingboot.com";
    
    private static String md5(String input) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] digest = md.digest(input.getBytes());
            StringBuilder sb = new StringBuilder();
            for (byte b : digest) {
                sb.append(String.format("x", b));
            }
            return sb.toString();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    
    private static String generateSign(long ts) {
        String firstMd5 = md5(APP_SECRET);
        String signRaw = firstMd5 + ts;
        return md5(signRaw);
    }
    
    public static String getMeteringData() throws Exception {
        long ts = System.currentTimeMillis() / 1000;
        String sign = generateSign(ts);
        
        String url = String.format("%s/%s/device/control/?sign=%s&ts=%d",
            API_BASE, APP_ID, sign, ts);
        
        String jsonBody = String.format(
            "{\"device\":\"%s\",\"order\":{\"metering\":\"\"}}",
            DEVICE_ID
        );
        
        HttpClient client = HttpClient.newHttpClient();
        HttpRequest request = HttpRequest.newBuilder()
            .uri(URI.create(url))
            .header("Content-Type", "application/json")
            .POST(HttpRequest.BodyPublishers.ofString(jsonBody))
            .build();
        
        HttpResponse<String> response = client.send(request,
            HttpResponse.BodyHandlers.ofString());
        return response.body();
    }
    
    public static void main(String[] args) throws Exception {
        String data = getMeteringData();
        System.out.println("计量数据: " + data);
    }
}

3.4 Shell脚本（cURL）实现

第四章：电量统计业务场景集成

4.1 系统集成架构

电量计量统计系统可与能源管理平台深度集成：

graph TB
    subgraph 硬件层
        A[智能大功率断路器40A
计量数显版]
    end
    
    subgraph 采集层
        B[定时采集服务]
        B1[实时功率]
        B2[累计电量]
        B3[电压/电流]
    end
    
    subgraph 存储层
        C[时序数据库
InfluxDB/TDengine]
        D[关系数据库
MySQL/SQLite]
    end
    
    subgraph 应用层
        E[能源看板]
        F[费用结算]
        G[异常告警]
        H[报表分析]
    end
    
    A -->|HTTP API| B
    B --> B1
    B --> B2
    B --> B3
    B --> C
    B --> D
    C --> E
    C --> H
    D --> F
    B1 --> G

4.2 第一种场景：实时功率监控

业务需求：实时监控设备/线路的当前功率，发现异常波动及时告警。

实现逻辑

定时采集当前功率数据
与预设阈值对比
超阈值时触发告警

代码实现

4.3 第二种场景：用电量统计与报表

业务需求：按日、周、月统计用电量，生成能耗报表。

实现逻辑

定时采集累计用电量
计算时间段内的用电差值
生成统计报表

代码实现

class EnergyReport:
    """能耗报表生成器"""
    
    def __init__(self, meter: EnergyMeter):
        self.meter = meter
    
    def get_daily_report(self, date: date) -> dict:
        """生成日报"""
        start = datetime.combine(date, datetime.min.time())
        end = datetime.combine(date, datetime.max.time())
        
        energy = self.meter.get_energy_consumption(str(start), str(end))
        
        # 获取该时段的最高功率
        conn = sqlite3.connect(self.meter.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            SELECT MAX(power), AVG(power), COUNT(*) 
            FROM metering_data 
            WHERE DATE(timestamp) = ?
        ''', (date,))
        max_power, avg_power, record_count = cursor.fetchone()
        conn.close()
        
        return {
            "date": str(date),
            "total_energy_kwh": round(energy, 2),
            "max_power_w": max_power or 0,
            "avg_power_w": round(avg_power or 0, 2),
            "sample_count": record_count or 0
        }
    
    def get_weekly_report(self, end_date: date = None) -> dict:
        """生成周报"""
        if end_date is None:
            end_date = date.today()
        start_date = end_date - timedelta(days=7)
        
        energy = self.meter.get_energy_consumption(str(start_date), str(end_date))
        
        return {
            "start_date": str(start_date),
            "end_date": str(end_date),
            "total_energy_kwh": round(energy, 2),
            "daily_avg_kwh": round(energy / 7, 2)
        }

4.4 第三种场景：共享充电计费结算

业务需求：在共享充电场景中，根据实际用电量进行费用结算。

实现逻辑

充电开始前记录初始累计电量
充电结束后再次查询累计电量
计算用电量差值，按电价计费

代码实现

class ChargingSession:
    """共享充电会话"""
    
    def __init__(self, session_id: str, price_per_kwh: float = 1.5):
        self.session_id = session_id
        self.price_per_kwh = price_per_kwh
        self.start_energy = None
        self.start_time = None
    
    def start_charging(self):
        """开始充电:记录初始电量"""
        data = get_metering_data()
        self.start_energy = data.get("energy", 0)
        self.start_time = datetime.now()
        set_power(True)  # 接通电源
        
        print(f"充电开始 - 初始电量: {self.start_energy}kWh")
        return {"session_id": self.session_id, "start_energy": self.start_energy}
    
    def end_charging(self) -> dict:
        """结束充电:计算费用"""
        data = get_metering_data()
        end_energy = data.get("energy", 0)
        end_time = datetime.now()
        set_power(False)  # 断开电源
        
        energy_used = end_energy - self.start_energy
        cost = energy_used * self.price_per_kwh
        duration = (end_time - self.start_time).total_seconds() / 3600
        
        result = {
            "session_id": self.session_id,
            "start_energy": self.start_energy,
            "end_energy": end_energy,
            "energy_used_kwh": round(energy_used, 3),
            "duration_hours": round(duration, 2),
            "cost_yuan": round(cost, 2)
        }
        
        print(f"充电结束 - 用电量: {energy_used}kWh, 费用: {cost}元")
        return result

4.5 场景四：用电异常检测

业务需求：检测用电异常（如夜间异常耗电、功率突变），及时发现设备故障或偷电行为。

实现逻辑

建立正常用电时段模型
实时对比实际用电与预期
偏差超阈值时触发告警

第五章：数据库设计与数据持久化

5.1 数据表设计

-- 设备配置表
CREATE TABLE energy_devices (
    id INT PRIMARY KEY AUTO_INCREMENT,
    device_id VARCHAR(20) NOT NULL UNIQUE,
    device_name VARCHAR(50),
    location VARCHAR(200),
    rated_power INT DEFAULT 8000,
    status TINYINT DEFAULT 1,
    created_at DATETIME DEFAULT NOW()
);

-- 计量数据明细表
CREATE TABLE metering_data (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    device_id VARCHAR(20) NOT NULL,
    timestamp DATETIME NOT NULL,
    power DECIMAL(10,2),      -- 当前功率(W)
    energy DECIMAL(10,3),     -- 累计用电量(kWh)
    voltage DECIMAL(8,2),     -- 电压(V)
    current DECIMAL(8,2),     -- 电流(A)
    INDEX idx_device_time (device_id, timestamp)
);

-- 日统计表
CREATE TABLE daily_energy_stats (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    device_id VARCHAR(20) NOT NULL,
    stat_date DATE NOT NULL,
    total_energy DECIMAL(10,3) DEFAULT 0,   -- 日用电量(kWh)
    max_power DECIMAL(10,2) DEFAULT 0,       -- 日最大功率(W)
    avg_power DECIMAL(10,2) DEFAULT 0,       -- 日平均功率(W)
    peak_hour VARCHAR(10),                   -- 峰值时段
    record_count INT DEFAULT 0,
    UNIQUE KEY uk_device_date (device_id, stat_date)
);

-- 月统计表
CREATE TABLE monthly_energy_stats (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    device_id VARCHAR(20) NOT NULL,
    stat_year INT NOT NULL,
    stat_month INT NOT NULL,
    total_energy DECIMAL(10,3) DEFAULT 0,
    max_daily_energy DECIMAL(10,3) DEFAULT 0,
    UNIQUE KEY uk_device_year_month (device_id, stat_year, stat_month)
);

-- 告警记录表
CREATE TABLE energy_alerts (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    device_id VARCHAR(20) NOT NULL,
    alert_type VARCHAR(50),     -- overload/under_voltage/surge/night_consumption
    alert_level TINYINT,         -- 1-提示 2-警告 3-严重
    power_value DECIMAL(10,2),
    threshold_value DECIMAL(10,2),
    message TEXT,
    is_resolved TINYINT DEFAULT 0,
    created_at DATETIME DEFAULT NOW()
);

5.2 数据采集最佳实践

第六章：故障排查与最佳实践

6.1 常见问题排查表

现象	可能原因	解决方案
设备显示离线	1. 电源断开2. WiFi密码错误/信号弱3. 路由器禁用了设备MAC	1. 检查AC供电2. 重配网络或靠近路由器3. 检查路由器白名单
计量查询失败	1. 命令格式错误2. 设备不支持计量功能3. 设备离线	1. 确认使用`{"metering":""}`格式2. 确认产品为计量版3.

第一章：场景痛点与解决概述

1.1 电量计量统计的业务需求

1.2 解决方案：计量数显版断路器实现电量统计

1.3 智能大功率断路器[计量数显版]40A核心参数

第二章：HTTP接口协议详解

2.1 整体架构

2.2 请求地址格式

2.3 签名算法（核心安全机制）

2.4 核心命令集

2.5 计量数据查询（核心功能）

2.6 完整请求示例

第三章：多语言代码实现

3.1 Python完整实现

3.2 Node.js实现

3.3 Java实现

3.4 Shell脚本（cURL）实现

第四章：电量统计业务场景集成

4.1 系统集成架构

4.2 第一种场景：实时功率监控

4.3 第二种场景：用电量统计与报表

4.4 第三种场景：共享充电计费结算

4.5 场景四：用电异常检测

第五章：数据库设计与数据持久化

5.1 数据表设计

5.2 数据采集最佳实践

第六章：故障排查与最佳实践

6.1 常见问题排查表

断路器产品方案：

门店电路控制：如何把35A智能空开开关接入到项目中

怎样二次开发25A远程智能空开以实现导轨式安装通断控制

实验室电源管理：怎么将智能断路器35A接入到自己的项目中

共享自习室电源控制：怎么将60A带计量额定 13200W 断路器对接到软件项目中

自助洗车设备电源控制：如何把35A定时控制断路器对接到自己的项目中

电量用途方案：

怎么二次开发50A计量版空开来实现实时电量计量统计

如何二次开发50A带计量远程控制断路器来实现实时电量计量统计

怎么在自助打印设备电源控制中集成智能设备来实现实时电量计量统计

如何对接50A导轨式计量断路器来实现实时电量计量统计

如何对接40A带计量数显智能电源控制来实现实时电量计量统计