怎么二次开发50A导轨式智能断路器来实现漏电保护控制_解决方案

CATALOG

解决方案：基于芯步开放接口的50A导轨式智能断路器二次开发——实现漏电保护控制

1. 我们需要解决什么问题？

首先，让我们明确一下目标。

你手里的这款50A导轨式智能断路器（型号UNI-DLQ-M-50A），本身是一个硬件设备。它现在能通过WiFi联网，并且芯步给它开放了一套HTTP API接口。

我们的核心任务是：不让断路器只当一个听命令的“开关”，而是要让它变成一个能自己判断漏电、自己跳闸的“保安”。

要实现这个，我们需要做一个“中间人”（也就是你的二次开发程序）。这个程序负责：

“监听”：不断地问断路器：你现在电流多少？有没有漏电？
“判断”：拿回来的数据如果超过了安全值（比如漏电30mA），立马判定有危险。
“执行”：立刻发指令回去：“马上断电！”

说白了，就是用软件逻辑给物理硬件加上一个‘智能大脑’。

2. 我们有什么可用资源？

在开始写代码之前，先点一下家底。芯步的开放接口给我们提供了什么？主要是两点

控制接口：用来发命令，比如“合闸”、“分闸”。这是我们执行“断电”的工具。
状态查询接口：用来问设备“你现在怎么样了？”，比如开关是开是关，电流参数是多少。这是我们做“漏电判断”的依据。

根据资料，这款设备支持标准的HTTP请求。你用什么语言都行，Java、Python、PHP、Go都支持。

特别提醒一句：一定要去芯步的控制台拿到你的 AppID、AppSecret 和 Device ID。这些都是调用接口的“身份证”，签名算法一般是 md5(md5(AppSecret) + ts)。

3. 硬核方案：如何一步步实现漏电保护？

既然是解决方案，咱们得来点实际的。这里，我设计了一套“主动轮询 + 阈值触发 + 自动重合闸”的方案。之所以不依赖设备主动上报（如果没做的话），是我们主动去问，这样最稳当。

第一步：获取实时电参（核心）

断路器的说明书里提到了支持命令，虽然直接列出的有 power（通断）等，但我们做漏电保护，更关心的是内部电量参数。理论上，通过对应的API（如设备状态接口），你可以获取到芯片采集的数据。

我们需要关心的核心数据有：

漏电电流：这是最重要的。一旦这个值不为0，就说明有电流“跑”到地上去了。
负载电流：看看现在总功率多大，防止过载。
开关状态：现在是合闸还是分闸状态。

怎么拿数据？一般的流程是：构造一个GET请求，带上签名，去请求设备状态的接口。

注意：如果官方文档没有直接的“读取漏电值”接口，就需要了解一下这款断路器是否支持在order指令中传递参数来返回状态。如果都不行，最直接的方法是利用定时任务配合功率变化来判断。比如，明明没发指令，功率突然归零了，那可能就是物理脱扣或者漏电跳闸了。

第二步：编写核心守护程序（Python伪代码示例）

为了让你看得更明白，我用Python写一段逻辑。这是一个运行在你的服务器或者云函数里的脚本。

# 引入需要的库
import requests
import time
import hashlib

# --- 配置区域（需要换成你自己的）---
AppID = "你的AppID"
AppSecret = "你的AppSecret"
DeviceID = "你的设备ID"
API_BASE_URL = "http://api.thingboot.com"  # 根据实际环境，如果是局域网可能地址不同

# 漏电保护阈值（单位:mA，一般人体安全阈值是30mA）
LEAKAGE_THRESHOLD = 30

# --- 核心函数:生成签名 ---
def generate_sign(ts):
    # 这里参考官方的加密逻辑:md5(md5(AppSecret) + ts)
    step1 = hashlib.md5(AppSecret.encode()).hexdigest()
    step2 = hashlib.md5((step1 + str(ts)).encode()).hexdigest()
    return step2

# --- 核心函数:获取设备状态（重点！）---
def get_device_status():
    ts = int(time.time())
    sign = generate_sign(ts)
    url = f"{API_BASE_URL}/{AppID}/device/status/"

params = {
        "device": DeviceID,
        "sign": sign,
        "ts": ts
    }
    try:
        response = requests.get(url, params=params, timeout=5)
        if response.status_code == 200:
            data = response.json()
            # 假设返回的json里有一个字段叫leakage_current，单位是mA
            # 具体字段名请查阅官方"产品手册"里的"功能列表"[citation:6]
            return {
                "status": data.get("status"),  # 0:关 1:开
                "leakage": data.get("leakage_current"), 
                "current": data.get("electric_current")
            }
        else:
            print(f"获取状态失败: {response.text}")
            return None
    except Exception as e:
        print(f"网络异常: {e}")
        return None

# --- 核心函数:执行分闸命令（跳闸）---
def trip_off():
    ts = int(time.time())
    sign = generate_sign(ts)
    url = f"{API_BASE_URL}/{AppID}/device/control/"
    
    # 关键指令:下发给设备，power=0 代表关闭输出
    payload = {
        "device": DeviceID,
        "order": {"power": 0}  # 官方文档中支持的命令格式[citation:1][citation:6]
    }
    headers = {'Content-Type': 'application/json'}
    
    params = {
        "sign": sign,
        "ts": ts
    }
    
    try:
        response = requests.post(url, params=params, json=payload, headers=headers, timeout=5)
        if response.status_code == 200:
            print("【执行成功】已发送断电指令！")
            return True
        else:
            print(f"【执行失败】{response.text}")
            return False
    except Exception as e:
        print(f"网络异常: {e}")
        return False

# --- 主循环:检测逻辑 ---
if __name__ == "__main__":
    print("启动漏电保护监控程序...")
    while True:
        # 1. 拿到当前数据
        device_data = get_device_status()
        
        if device_data and device_data["leakage"] is not None:
            # 2. 判断是否有漏电
            if device_data["leakage"] > LEAKAGE_THRESHOLD:
                print(f"⚠️ 危险！检测到漏电电流 {device_data['leakage']}mA，超过阈值 {LEAKAGE_THRESHOLD}mA！")
                
                # 3. 执行断电
                trip_off()
                
                # 4. 记录日志，可以插入数据库或者发告警给管理员
                print("记录告警:设备已跳闸，请检查线路。")
                
                # 5. 防止频繁重启，休眠一段时间
                time.sleep(10)
            else:
                # 一切正常
                print(f"正常监控中... 当前漏电: {device_data['leakage']}mA")
        else:
            print("等待设备上线...")
        
        # 轮询间隔:考虑到漏电的及时性， 0.5秒 - 1秒 检测一次[citation:8]
        time.sleep(0.5)

第三步：针对“漏电保护”的特殊处理（进阶）

上面的代码只是一个基础模型。在实际工程中，漏电保护有几个细节要把握好：

防抖动（确认动作）有时候电网波动，互感器会误报一瞬间的漏电。我们不能一有数据就跳闸。优化方法：连续检测3次，如果每次都超过阈值，且持续时间超过20毫秒，再执行跳闸。
重合闸逻辑对于临时性漏电（比如大功率设备启动瞬间），如果直接跳闸导致停电，体验很差。可以在程序中设置一个重合闸机制
- 跳闸后，等待5秒。
- 再次检测漏电值。
- 如果漏电消失，自动发指令 {"power": 1} 尝试合闸。
- 如果合闸瞬间又立刻触发漏电，那就锁定不再合闸，必须报修。
控制响应速度官方数据显示，从命令下发到设备执行大概是80-120毫秒。加上网络来回，你的程序必须运行在离云服务器近的地方（比如用云函数），或者直接用局域网API（私有化部署），这样才能把总响应时间控制在0.3秒以内。

4. 本地局域网部署（进阶方案）

如果你是工厂或者机房使用，不希望数据经过外网，担心延迟或断网。芯步的设备是支持局域网API的。在这种情况下，你的程序不需要经过公网云平台，直接在局域网里通过HTTP请求就可以控制断路器。这能保证即便外网断了，你的漏电保护逻辑依然有效，延迟也会更低。

5. 总结

通过芯步的开放接口二次开发实现漏电保护，核心思路就是“高频轮询 + 即时控制”。

物理基础：利用断路器本身的电流采样能力。
传输通道：利用开放的HTTP API。
核心逻辑：通过你写的Python/Java程序，用代码定义漏电阈值和保护逻辑。

这样一来，你就不再受限于厂商App的功能了。你可以把这套逻辑集成到自己的工厂MES系统、智慧楼宇系统，甚至是家里的HomeAssistant里，实现真正的定制化安全控制。