怎么二次开发智能通断器 AC5-10A以实现能耗数据统计控制_解决方案

CATALOG

一、写在前面：这东西能干嘛？

咱们先不急着敲代码，聊聊这个智能通断器AC5-10A到底是个啥。说白了，它就是一个可以远程控制电路通断的“智能开关”，还能告诉你用了多少电。

核心卖点

一路继电器控制，额定电流10A，最高2200W阻性负载（普通电器没问题，电机之类感性负载要降额到350W以下）
直接连WiFi 2.4G，不需要额外买网关，插电就能用
支持HTTP接口调用，意味着你用任何能发HTTP请求的语言都能控制它——Python、Java、C#、Node.js，甚至命令行curl都行

适用场景：宿舍限电管控、机房设备远程重启、充电桩计费、智能家居能耗统计……总之就是需要“遥控开关+知道用了多少电”的地方。

二、准备工作：你需要拿到三样东西

在动手之前，先把这几样东西准备好：

项目	说明	去哪找
AppID	你的应用标识	登录芯步控制台 → 开发设置
AppSecret	应用密钥（别泄露！）	同上
Device ID	设备的唯一ID	设备外壳标签 / 控制台设备列表

这三个值是所有API调用的“身份证”，没有它们玩不转。

另外注意：AC5-10A这个型号支不支持电量采集？ 我查到的资料里，芯步的这款产品主打通断控制和HTTP接口开放，但具体的电流/电压/功率数据采集能力在产品手册中没有明确列出。如果你的核心需求是采集实时电量数据，：

先确认一下你手里的设备版本是否带“计量款”后缀
或者直接联系芯步的技术支持确认

一个小提示：同样做智能通断器的智鸟科技，明确区分了“常规款”（只有通断控制）和“采集款”（带电压/电流/功率/电量统计）。芯步的AC5-10A如果也是类似的逻辑，那做能耗统计就需要计量款。本文的方案假设设备支持电量数据上报，如果实际不支持，可以调整侧重点为“基于通断次数/时长的粗粒度能耗估算”。

三、核心接口详解：怎么控制和取数据

芯步的开放接口分为下行（你→设备）和上行（设备→你）两类。做能耗统计，两类都要用。

3.1 控制设备通断（下行）

这是最基础的功能——远程开关。

请求地址

https://api.thingboot.com/{AppID}/device/control/?sign={签名}&ts={时间戳}

请求方式：POST（推荐，参数长了用GET不方便）

核心参数

device：设备ID，就是上面说的那个
order：命令内容，JSON字符串格式

开关命令示例

具体的命令名要看产品手册，可能是power、power1、key等。AC5-10A只有一路，大概率就是power。

3.2 获取能耗数据（上行）

能耗统计的关键——设备主动上报数据。

芯步的机制是：设备状态变化（如开关动作、定时上报）时，云平台会向你自己配置的接收地址（URL）推送消息。

你需要做的

在控制台设置一个接收消息的URL（就是你服务器的某个接口地址）
当设备上报数据时，芯步会POST一个JSON到你配置的地址
你的服务器接收、解析、存数据库

上报数据大概长这样（根据同类产品推测）：

⚠️ 关键提醒：AC5-10A具体上报哪些字段，一定要去看它的《产品手册》。计量类设备一般会上报电压/电流/功率/累计电量，但如果你的设备不是计量款，这些字段可能压根不存在。

四、代码实操：用Python快速实现

下面用Python写一个完整的demo，包含签名生成、下发开关命令、接收能耗上报三个核心功能。

4.1 签名生成（所有接口通用的认证方式）

芯步的签名算法是md5(md5(AppSecret) + ts)，稍微有点绕

import hashlib
import time
import requests
import json

def generate_sign(app_secret, ts):
    """
    生成芯步API签名
    算法:md5(md5(AppSecret) + ts)
    """
    # 第一次MD5
    secret_md5 = hashlib.md5(app_secret.encode()).hexdigest()
    # 拼接时间戳
    sign_str = secret_md5 + str(ts)
    # 第二次MD5
    sign = hashlib.md5(sign_str.encode()).hexdigest()
    return sign

def control_device(app_id, app_secret, device_id, power_state):
    """
    控制设备通断
    power_state: 1=开, 0=关
    """
    ts = int(time.time())
    sign = generate_sign(app_secret, ts)
    
    url = f"https://api.thingboot.com/{app_id}/device/control/"
    params = {
        "sign": sign,
        "ts": ts
    }
    # 注意order要传JSON字符串
    payload = {
        "device": device_id,
        "order": json.dumps({"power": power_state})
    }
    
    response = requests.post(url, params=params, json=payload)
    return response.json()

# 使用示例
if __name__ == "__main__":
    APP_ID = "你的AppID"
    APP_SECRET = "你的AppSecret"
    DEVICE_ID = "你的设备ID"
    
    # 打开设备
    result = control_device(APP_ID, APP_SECRET, DEVICE_ID, 1)
    print(result)  # 正常返回 {"code": 200}

4.2 接收能耗上报（Flask服务）

你需要跑一个Web服务来接收设备推送的数据：

from flask import Flask, request, jsonify
import json
import sqlite3
from datetime import datetime

app = Flask(__name__)

# 初始化数据库
def init_db():
    conn = sqlite3.connect('energy_data.db')
    c = conn.cursor()
    c.execute('''
        CREATE TABLE IF NOT EXISTS energy_records (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            device_id TEXT,
            timestamp INTEGER,
            power_state INTEGER,
            voltage REAL,
            current REAL,
            power_w REAL,
            energy_kwh REAL,
            receive_time TEXT
        )
    ''')
    conn.commit()
    conn.close()

@app.route('/webhook/device/report', methods=['POST'])
def handle_device_report():
    """
    芯步推送的设备上报数据接收接口
    把这个URL填到控制台的"消息接收地址"里
    """
    data = request.get_json()
    print(f"收到上报: {json.dumps(data, indent=2)}")
    
    # 解析数据（具体字段参考产品手册）
    device_id = data.get('device_id')
    ts = data.get('timestamp')
    status = data.get('status', {})
    
    # 存数据库
    conn = sqlite3.connect('energy_data.db')
    c = conn.cursor()
    c.execute('''
        INSERT INTO energy_records 
        (device_id, timestamp, power_state, voltage, current, power_w, energy_kwh, receive_time)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    ''', (
        device_id, ts,
        status.get('power'),
        status.get('voltage'),
        status.get('current'),
        status.get('power_w'),
        status.get('energy_kwh'),
        datetime.now().isoformat()
    ))
    conn.commit()
    conn.close()
    
    # 芯步要求返回特定格式确认收到
    return jsonify({"code": 200, "msg": "ok"})

if __name__ == '__main__':
    init_db()
    app.run(host='0.0.0.0', port=8000)

把这个服务跑起来之后，记下外网可访问的地址（比如https://your-server.com/webhook/device/report），然后去芯步控制台配置消息推送地址。

4.3 查询统计数据

数据存下来了，怎么查？写几个SQL：

def get_daily_energy(device_id, date_str):
    """
    查询某天的用电量
    date_str格式: '2024-01-15'
    """
    conn = sqlite3.connect('energy_data.db')
    c = conn.cursor()
    # 假设我们记录的是累计电量，取当天第一条和最后一条的差值
    c.execute('''
        SELECT energy_kwh, timestamp FROM energy_records 
        WHERE device_id = ? AND date(receive_time) = ?
        ORDER BY timestamp ASC
    ''', (device_id, date_str))
    rows = c.fetchall()
    conn.close()
    
    if len(rows) >= 2:
        start_energy = rows[0][0]
        end_energy = rows[-1][0]
        return end_energy - start_energy
    return 0

def get_current_power(device_id):
    """获取实时功率"""
    conn = sqlite3.connect('energy_data.db')
    c = conn.cursor()
    c.execute('''
        SELECT power_w, receive_time FROM energy_records 
        WHERE device_id = ? AND power_w IS NOT NULL
        ORDER BY receive_time DESC LIMIT 1
    ''', (device_id,))
    row = c.fetchone()
    conn.close()
    return {"power_w": row[0], "time": row[1]} if row else None

五、进阶玩法：加上智能控制逻辑

有了通断控制和能耗数据，就可以搞点“自动化的骚操作”了。

5.1 过载自动断电保护

5.2 定时任务 + 电量统计

比如每天晚上统计当天用电，如果超过设定值就发告警：

5.3 注意：数据上报频率问题

从同类设备的经验来看（比如Sonoff的某些型号），这类设备可能默认一天只上报一次数据（比如晚上11点），只有在主动通过APP查看或设备重启时才实时上报。

这意味着：

如果你想做实时功率监控，可能需要确认设备是否支持高频上报，或者在每次开关动作时主动查询状态
如果设备支持设置上报间隔（查一下产品手册），可以调成1分钟或5分钟一次

六、常见坑和避坑指南

根据我整理资料时看到的用户反馈，列几个容易踩的坑：

坑	表现	解决方案
电量数据不更新	数据库里能耗一直是同一个值	确认设备是否支持计量；检查上报频率设置；尝试在APP里主动刷新触发上报
命令下发成功但设备没反应	API返回200，但灯不亮	设备可能离线了，检查WiFi连接；确认`order`里的命令名是否正确（是`power`还是`power1`？）
收不到上报推送	服务器没收到POST请求	检查你的URL是否公网可访问（内网穿透用ngrok）；确认控制台配置的地址正确
签名错误	返回401之类	检查AppSecret对不对；确认ts是秒级时间戳；签名算法是否写错了

另外，关于私有化部署：芯步支持自建消息服务器，也就是说如果你不想走他们的云平台，可以自己搭MQTT或HTTP服务器，设备直连你的服务器。这个适合内网环境或者对数据隐私要求高的场景。

七、总结

把整个流程串起来，大概就这几步：

拿资料：AppID、AppSecret、Device ID，翻产品手册确认电量字段
搭接收服务：跑一个HTTP服务等着收设备上报的数据
写控制逻辑：封装开关命令、签名生成
存数据+做统计：数据库存上报记录，写SQL算用电量
加自动化：过载保护、定时报告、异常告警……

芯步的这套接口设计得还算友好，核心就是HTTP + 签名 + JSON命令，三件套搞定。开发过程中如果遇到问题，多看产品手册和接口文档，或者直接找芯步的技术支持——毕竟硬件厂家最清楚自己设备上报什么数据。

最后提醒一句：AC5-10A的额定电流是10A，接大功率电器（空调、热水器）时注意别超了，尤其感性负载要降额使用。安全第一，代码第二！