怎么二次开发20A远程控制断路器来实现漏电保护控制_解决方案

CATALOG

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

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

1.1 漏电保护的挑战

在智能配电系统中，漏电保护是保障人身安全的核心功能。传统漏电保护方案存在以下痛点：

痛点	具体表现	带来的问题
保护功能单一	仅能检测漏电跳闸，无数据记录	无法分析漏电规律和原因
状态不可知	跳闸后需现场排查状态	故障恢复时间长
无法远程复位	跳闸后必须人工到场合闸	无人值守场景下业务中断
缺乏智能分析	无法区分突变/缓变漏电	次生误跳影响正常用电

1.2 解决方案：软件二次开发实现智能漏电保护

通过将芯步20A远程控制断路器集成到配电管理软件中，并进行二次开发，实现“漏电检测-联动跳闸-远程告警-智能复位”的完整保护闭环。

芯步20A智能断路器核心参数

参数项	规格详情
产品型号	UNI-DLQ-20A
联网方式	WiFi 2.4GHz / 4G可选
额定电流	MAX 20A
阻性负载	MAX 4000W
感性负载	MAX 600W（电机/空调等需降额）
工作电压	AC 85-265V
安装方式	35mm DIN导轨式
外壳材质	阻燃V0级PC塑料

核心要点：该设备原生支持HTTP接口远程控制，但不内置漏电检测传感器。因此，实现漏电保护控制的二次开发，核心是软件系统+外置漏电检测传感器+断路器的联动方案。

第二章：漏电保护原理与二次开发架构

2.1 智能漏电保护的工作原理

一个完整的智能漏电保护系统，其工作流程如下：

零序电流互感器检测漏电 → 信号调理与放大 → ADC采样 → 
软件算法判断（突变/缓变） → 触发控制指令 → 
断路器执行跳闸 → 上位机记录与告警

在智能漏电断路器设计中，主程序需遵循高优先级任务（漏电采样）>中优先级任务（状态显示）>低优先级任务（通信处理）的任务调度原则。

2.2 二次开发整体架构

由于芯步20A断路器本身不内置漏电检测模块，二次开发的核心是构建“漏电检测传感器 + 软件判断逻辑 + 断路器联动”的三层架构

graph TB
    subgraph 硬件层
        A[零序电流互感器
漏电检测]
        B[信号调理电路
放大/滤波]
        C[ADC采样模块
单片机/采集卡]
        D[芯步20A断路器]
    end
    
    subgraph 软件层
        E[漏电数据采集服务]
        F[漏电判断算法]
        G[断路器控制模块]
        H[告警与记录模块]
    end
    
    subgraph 应用层
        I[管理后台/SCADA]
        J[短信/App推送]
        K[数据报表分析]
    end
    
    A --> B --> C --> E
    E --> F
    F -->|超阈值| G
    F -->|记录| H
    G -->|HTTP API| D
    H --> I
    H --> J
    H --> K

2.3 漏电判断算法设计

漏电保护的核心算法需要区分两种漏电类型

漏电类型	特征	处理策略
突变漏电	电流上升率 > 500mA/ms	立即触发脱扣（0.1s内）
缓变漏电	电流持续超阈值但上升缓慢	按预设延时动作（0.3s/0.6s），避免误动

漏电动作阈值参考（GB 16917标准）：

一般场所：30mA
潮湿场所：15mA
设备保护：100mA

2.4 断路器控制命令

芯步20A断路器支持以下核心命令

命令	格式	说明
接通	`{"power1":"1"}`	合闸通电
断开	`{"power1":"0"}`	跳闸断电
先断后通	`{"reset1":"3000"}`	断电3秒后自动恢复
先通后断	`{"point1":"3000"}`	通电3秒后自动断开
状态查询	`{"get_status":""}`	查询当前通断状态

注意：命令中的power1表示第1路输出，不同产品版本可能使用power或power1，请以产品手册为准。

第三章：硬件选型与部署方案

3.1 漏电检测传感器选型

要实现漏电保护，需要额外配置漏电检测传感器：

方案	传感器类型	输出信号	适用场景
方案一	零序电流互感器 + 信号调理模块	模拟量（0-5V/4-20mA）	标准配电场景
方案二	集成式漏电检测模块	开关量（干接点）	快速集成场景
方案三	智能电表+漏电检测	数字通信（Modbus）	需要更多电力参数

推荐方案一：采用零序电流互感器（如TA-30型）配合信号调理电路，输出0-5V模拟信号，精度高、可调阈值。

3.2 ADC采样硬件配置

将模拟信号转换为数字量供软件判断：

选项	设备	优点
低成本	Arduino Nano + ADC模块	开发简单，成本低
工业级	数据采集卡（DAQ）	精度高，抗干扰强
集成方案	带ADC的PLC	一体化，可靠性高

3.3 整体部署架构图

graph LR
    subgraph 配电箱内
        A[市电输入] --> B[芯步20A断路器]
        B --> C[负载]
        D[零序电流互感器] --> E[信号调理电路]
        E --> F[ADC采集模块]
    end
    
    subgraph 边缘侧
        F --> G[边缘计算网关
运行漏电判断软件]
        G -->|HTTP指令| B
    end
    
    subgraph 云端
        G --> H[芯步API/私有服务器]
        H --> I[管理后台]
    end

第四章：软件二次开发完整实现

4.1 整体设计

本二次开发方案包含三个核心模块：

漏电数据采集模块：读取ADC采样值，计算漏电电流
漏电判断与决策模块：实现突变/缓变判断算法，决定是否跳闸
断路器控制模块：调用芯步API执行通断控制

4.2 签名算法（与之前方案一致）

芯步API采用双重MD5签名机制：

Sign = md5( md5(AppSecret) + ts )

计算步骤

对AppSecret进行第一次MD5，得到32位小写字符串
将结果与时间戳ts（秒级）拼接
对拼接后的字符串进行第二次MD5

4.3 断路器控制核心代码

# -*- coding: utf-8 -*-
import hashlib
import time
import requests
import json
from typing import Dict, Optional
from datetime import datetime
import threading
import queue

# ========== 配置信息 ==========
APP_ID = "YOUR_APP_ID"
APP_SECRET = "YOUR_APP_SECRET"
BREAKER_DEVICE_ID = "YOUR_BREAKER_DEVICE_ID"  # 断路器设备ID
API_BASE = "https://api.thingboot.com"

# 漏电保护配置
LEAKAGE_THRESHOLD_MA = 30  # 漏电动作阈值(mA)，国标一般场所30mA
SUDDEN_CHANGE_RATE = 500   # 突变判断阈值(mA/ms)
GRADUAL_DELAY_MS = 300     # 缓变漏电延时动作时间(ms)
# =============================

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

def get_sign_and_ts() -> tuple:
    """生成签名和时间戳"""
    ts = int(time.time())
    md5_secret = calculate_md5(APP_SECRET)
    sign_raw = md5_secret + str(ts)
    sign = calculate_md5(sign_raw)
    return sign, ts

def send_breaker_command(order: dict) -> dict:
    """向断路器发送命令"""
    sign, ts = get_sign_and_ts()
    url = f"{API_BASE}/{APP_ID}/device/control/"
    params = {"sign": sign, "ts": ts}
    payload = {"device": BREAKER_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 breaker_trip():
    """断路器跳闸（断开）"""
    result = send_breaker_command({"power1": "0"})
    print(f"[{datetime.now()}] 断路器跳闸执行结果: {result}")
    return result

def breaker_close():
    """断路器合闸（接通）"""
    result = send_breaker_command({"power1": "1"})
    print(f"[{datetime.now()}] 断路器合闸执行结果: {result}")
    return result

def breaker_auto_reset(delay_ms: int = 30000):
    """延时自动复位:先断开，等待指定毫秒后自动恢复"""
    result = send_breaker_command({"reset1": str(delay_ms)})
    print(f"[{datetime.now()}] 断路器自动复位指令已发送，{delay_ms}ms后恢复")
    return result

def get_breaker_status() -> dict:
    """查询断路器当前状态"""
    return send_breaker_command({"get_status": ""})

4.4 漏电检测与判断模块

这是二次开发的核心：实现漏电数据采集和智能判断逻辑。

# ========== 漏电检测模块（需配合硬件ADC） ==========
import numpy as np
from collections import deque

class LeakageDetector:
    """漏电检测器"""
    
    def __init__(self, threshold_ma=30, sudden_rate=500, gradual_delay_ms=300):
        self.threshold_ma = threshold_ma          # 漏电阈值(mA)
        self.sudden_rate = sudden_rate            # 突变速率阈值(mA/ms)
        self.gradual_delay_ms = gradual_delay_ms  # 缓变延时(ms)
        
        # 数据缓存
        self.value_buffer = deque(maxlen=100)      # 历史值缓存
        self.last_value = 0
        self.last_time = 0
        self.gradual_start_time = None
        self.is_tripped = False
        
    def adc_to_current(self, adc_value: int, v_ref: float = 5.0, 
                       adc_max: int = 4095, ct_ratio: int = 1000) -> float:
        """
        将ADC采样值转换为漏电电流(mA)
        
        参数:
            adc_value: ADC原始值(0-4095)
            v_ref: 参考电压(V)
            adc_max: ADC最大值
            ct_ratio: 互感器变比(如1000:1表示原边1000A对应副边1A)
        """
        # ADC转电压
        voltage = (adc_value / adc_max) * v_ref
        # 电压转副边电流(假设信号调理电路转换系数)
        secondary_current = voltage / 0.1  # 假设0.1V对应1A
        # 副边电流转原边漏电电流(mA)
        primary_current_ma = secondary_current * (1000 / ct_ratio) * 1000
        return primary_current_ma
    
    def calculate_rate(self, current_ma: float, current_time_ms: int) -> float:
        """计算漏电电流变化率(mA/ms)"""
        if self.last_time == 0:
            return 0
        delta_value = current_ma - self.last_value
        delta_time = current_time_ms - self.last_time
        if delta_time == 0:
            return 0
        return delta_value / delta_time
    
    def detect(self, current_ma: float, current_time_ms: int) -> Dict:
        """
        漏电检测主函数
        
        返回:
            {
                "should_trip": bool,      # 是否应该跳闸
                "reason": str,            # 跳闸原因
                "current_ma": float,      # 当前漏电电流
                "rate": float             # 变化率
            }
        """
        # 计算变化率
        rate = self.calculate_rate(current_ma, current_time_ms)
        
        # 更新缓存
        self.value_buffer.append(current_ma)
        self.last_value = current_ma
        self.last_time = current_time_ms
        
        result = {
            "should_trip": False,
            "reason": "",
            "current_ma": current_ma,
            "rate": rate
        }
        
        # 1. 突变漏电检测:电流超过阈值且变化率超过突变阈值
        if current_ma >= self.threshold_ma and rate >= self.sudden_rate:
            result["should_trip"] = True
            result["reason"] = f"突变漏电保护: 电流{current_ma:.1f}mA, 变化率{rate:.1f}mA/ms"
            self.gradual_start_time = None
            
        # 2. 缓变漏电检测:电流超过阈值但变化率较低
        elif current_ma >= self.threshold_ma:
            if self.gradual_start_time is None:
                self.gradual_start_time = current_time_ms
            elapsed = current_time_ms - self.gradual_start_time
            if elapsed >= self.gradual_delay_ms:
                result["should_trip"] = True
                result["reason"] = f"缓变漏电保护: 电流{current_ma:.1f}mA持续{elapsed}ms"
                self.gradual_start_time = None
        else:
            # 电流恢复正常，重置缓变计时
            self.gradual_start_time = None
            
        return result
    
    def reset(self):
        """重置检测器状态（跳闸复位后调用）"""
        self.is_tripped = False
        self.gradual_start_time = None
        self.value_buffer.clear()
        self.last_value = 0
        self.last_time = 0

4.5 完整漏电保护服务

# ========== 漏电保护主服务 ==========
import threading
import time
import schedule

class LeakageProtectionService:
    """漏电保护服务"""
    
    def __init__(self, adc_read_func, sampling_interval_ms=10):
        """
        参数:
            adc_read_func: ADC读取函数，返回(adc_value, timestamp_ms)
            sampling_interval_ms: 采样间隔(ms)
        """
        self.detector = LeakageDetector()
        self.adc_read_func = adc_read_func
        self.sampling_interval_ms = sampling_interval_ms
        self.running = False
        self.trip_logs = []  # 跳闸记录
        self.auto_reclose_enabled = False  # 是否启用自动重合闸
        self.reclose_delay_ms = 30000      # 重合闸延时(ms)
        
    def start(self):
        """启动漏电保护服务"""
        self.running = True
        thread = threading.Thread(target=self._monitor_loop)
        thread.daemon = True
        thread.start()
        print("[服务] 漏电保护服务已启动")
        
    def stop(self):
        """停止服务"""
        self.running = False
        print("[服务] 漏电保护服务已停止")
        
    def _monitor_loop(self):
        """监控主循环"""
        while self.running:
            try:
                # 读取ADC值
                adc_value, timestamp_ms = self.adc_read_func()
                
                # 转换为漏电电流(mA)
                current_ma = self.detector.adc_to_current(adc_value)
                
                # 漏电判断
                result = self.detector.detect(current_ma, timestamp_ms)
                
                # 记录日志
                if current_ma > 5:  # 超过5mA记录
                    self._log_sample(current_ma, result["rate"])
                
                # 触发跳闸
                if result["should_trip"]:
                    self._handle_trip(result)
                    
            except Exception as e:
                print(f"[错误] 监控循环异常: {e}")
                
            time.sleep(self.sampling_interval_ms / 1000.0)
    
    def _handle_trip(self, result: dict):
        """处理跳闸事件"""
        print(f"[跳闸] {result['reason']}")
        
        # 记录跳闸事件
        trip_record = {
            "timestamp": datetime.now().isoformat(),
            "current_ma": result["current_ma"],
            "rate": result["rate"],
            "reason": result["reason"]
        }
        self.trip_logs.append(trip_record)
        
        # 执行跳闸
        trip_result = breaker_trip()
        
        # 发送告警
        self._send_alert(trip_record)
        
        # 自动重合闸逻辑
        if self.auto_reclose_enabled:
            self._schedule_reclose()
    
    def _schedule_reclose(self):
        """调度自动重合闸"""
        def do_reclose():
            # 先检查漏电是否已消除
            adc_value, _ = self.adc_read_func()
            current_ma = self.detector.adc_to_current(adc_value)
            if current_ma < self.detector.threshold_ma:
                breaker_close()
                self.detector.reset()
                print("[重合闸] 漏电已消除，自动合闸成功")
                self._send_notification("自动重合闸成功")
            else:
                print(f"[重合闸] 漏电未消除(当前{current_ma:.1f}mA)，禁止合闸")
                self._send_alert("自动重合闸失败，漏电未消除")
                
        timer = threading.Timer(self.reclose_delay_ms / 1000.0, do_reclose)
        timer.daemon = True
        timer.start()
        
    def _log_sample(self, current_ma: float, rate: float):
        """记录采样数据（可选:写入数据库）"""
        pass  # 实际项目中写入时序数据库
        
    def _send_alert(self, record: dict):
        """发送告警"""
        message = f"【漏电保护】{record['reason']}，断路器已跳闸"
        # 实际项目中调用短信/邮件/钉钉接口
        print(f"[告警] {message}")
        
    def _send_notification(self, message: str):
        """发送通知"""
        print(f"[通知] {message}")
        
    def get_trip_history(self, limit=100):
        """获取跳闸历史"""
        return self.trip_logs[-limit:]

# ========== ADC读取示例（模拟） ==========
def mock_adc_read():
    """模拟ADC读取函数（实际项目中替换为真实硬件读取）"""
    import random
    # 模拟漏电电流0-50mA随机波动
    import math
    # 实际应读取真实ADC值
    adc_value = random.randint(0, 500)  # 模拟值
    timestamp_ms = int(time.time() * 1000)
    return adc_value, timestamp_ms

# ========== 启动服务 ==========
if __name__ == "__main__":
    # 创建漏电保护服务
    service = LeakageProtectionService(mock_adc_read, sampling_interval_ms=10)
    
    # 配置自动重合闸（可选）
    service.auto_reclose_enabled = True
    service.reclose_delay_ms = 30000  # 30秒后尝试重合闸
    
    # 启动服务
    service.start()
    
    # 保持主线程运行
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        service.stop()

4.6 状态查询与远程复位API

# ========== 用于管理后台的API接口（Flask示例） ==========
from flask import Flask, request, jsonify

app = Flask(__name__)
protection_service = None  # 全局服务实例

@app.route('/api/breaker/status', methods=['GET'])
def get_status():
    """获取断路器状态和漏电信息"""
    breaker_status = get_breaker_status()
    return jsonify({
        "breaker": breaker_status,
        "leakage_protection_active": protection_service.running if protection_service else False
    })

@app.route('/api/breaker/close', methods=['POST'])
def manual_close():
    """手动合闸（需管理员权限）"""
    # 先检查当前漏电状态
    adc_value, _ = mock_adc_read()  # 实际读取真实ADC
    detector = LeakageDetector()
    current_ma = detector.adc_to_current(adc_value)
    
    if current_ma >= 30:
        return jsonify({"success": False, "reason": f"当前漏电电流{current_ma:.1f}mA，禁止合闸"}), 403
    
    result = breaker_close()
    return jsonify({"success": True, "result": result})

@app.route('/api/breaker/trip', methods=['POST'])
def manual_trip():
    """手动跳闸"""
    result = breaker_trip()
    return jsonify({"success": True, "result": result})

@app.route('/api/leakage/history', methods=['GET'])
def get_leakage_history():
    """获取跳闸历史记录"""
    if protection_service:
        history = protection_service.get_trip_history()
        return jsonify({"history": history})
    return jsonify({"history": []})

第五章：二次开发部署方案

5.1 边缘计算网关方案

推荐使用边缘计算网关来运行漏电保护服务，实现本地闭环控制：

网关选项	配置	优点
树莓派4B	4GB RAM, 32GB存储	开发灵活，成本低
工业网关	工业级工控机	可靠性高，适合工业环境
现有工控机	工厂现有PLC/工控机	复用现有资源

5.2 网络部署架构

graph TB
    subgraph 本地局域网
        A[漏电检测传感器] --> B[ADC采集模块]
        B --> C[边缘网关
运行漏电保护服务]
        C -->|HTTP| D[芯步断路器]
    end
    
    subgraph 云端/公网
        C -->|可选| E[芯步云平台]
        E --> F[管理后台]
    end
    
    C -->|告警| G[钉钉/企业微信]
    C -->|日志| H[本地数据库]

5.3 关键参数配置

参数	推荐值	说明
采样间隔	10ms	每10ms采样一次，满足实时性要求
漏电阈值	30mA	国标一般场所要求
突变速率阈值	500mA/ms	区分突变与缓变
缓变延时	300ms	避免误动作
自动重合闸延时	30s	故障消除后自动恢复
历史记录保留	1000条	便于故障分析

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

6.1 断路器控制常见问题

现象	可能原因	解决方案
设备离线	WiFi信号弱/电源断开	检查网络、缩短距离或换有线版
跳闸命令无响应	设备离线/命令格式错误	确认device_id正确，检查order格式
控制成功但无动作	接线错误/负载问题	检查L-IN/L-OUT接线是否正确

6.2 漏电检测常见问题

现象	可能原因	解决方案
误跳闸	阈值设置过低/谐波干扰	检查阈值30mA是否合理，加装滤波
漏检	互感器安装不当	确认零序互感器穿过了所有相线和零线
ADC读数异常	信号调理电路故障	检查放大电路供电和基准电压

6.3 代码层面注意事项

实时性保障：采样间隔≤10ms，漏电突变检测才能达到国标要求
防抖处理：对ADC采样值进行数字滤波（如移动平均），避免噪声误触发
心跳保活：定期查询设备在线状态，设备离线时应有本地告警
日志完整：记录每次跳闸的原因、电流值、时间戳，便于故障追溯

第七章：总结

通过本文的二次开发方案，将芯步20A远程控制断路器与外置漏电检测传感器、软件判断逻辑相结合，成功构建了完整的智能漏电保护系统。

核心要点回顾

硬件方案：20A断路器 + 零序电流互感器 + ADC采集模块
判断算法：区分突变漏电（立即跳闸）和缓变漏电（延时跳闸）
控制接口{"power1":"0"}跳闸，{"power1":"1"}合闸，{"reset1":"30000"}自动复位
部署架构：边缘网关本地闭环控制 + 云端可选管理
扩展能力：支持自动重合闸、历史记录分析、多渠道告警

对接工作量评估：硬件改造约1-2天（安装互感器+采集模块），软件开发约2-3天完成采样、判断、控制的完整闭环。

安全要点提醒

⚠️ 断路器接线前必须断电
⚠️ 感性负载（电机/空调）功率必须≤600W
⚠️ 漏电保护是安全设备，开发完成后必须进行功能验证测试

让每一路配电都拥有"智能感知+精准执行"的漏电保护能力——这是软件定义保护的核心价值，也是从"被动跳闸"到"主动防御"的升级路径。