怎样对接30A 智能开关以实现远程开关状态查询_解决方案

CATALOG

一、背景与需求分析

在工业设备控制、智能充电桩、大功率电器管理等场景中，除了远程通断控制，设备状态的实时感知同样是核心需求。传统方案面临以下痛点：

状态不可知：电路实际是通是断无法远程确认，运维人员只能依赖现场查看
故障定位困难：设备离线、跳闸、异常断电时无法第一时间获知原因
告警响应滞后：无法实现状态变化自动通知，重要事件依赖人工巡检
系统集成缺环：在自己开发的软件平台中，缺少统一的状态监控入口

芯步智能断路器30A（型号：UNI-DLQ-30A）提供了一套大功率承载、开放API、状态主动上报的解决方案。该产品支持30A额定电流（6600W负载），适用于工业设备、充电桩、大功率电器等场景；不仅支持HTTP远程通断控制，还具备设备状态实时上报能力，可通过消息推送机制将开关状态变化主动推送到自己的服务器。

本方案将从产品选型、状态查询架构、消息推送对接、API控制开发、典型场景配置五个维度，详细阐述如何对接30A智能开关并实现远程开关状态查询。

二、产品选型与技术规格

2.1 智能断路器30A核心参数

参数项	规格说明
产品型号	UNI-DLQ-30A
额定电流	MAX 30A
负载功率	6600W（适用于大功率设备）
无线连接	WiFi 2.4GHz IEEE 802.11 b/g/n
控制方式	HTTP接口远程控制 + 手动按钮
状态上报	设备状态变化自动推送到云端[MQTT]
私有化部署	支持自建消息服务器，可在纯局域网环境运行
网络备份	可设定5组WiFi网络，优先连接信号最强的网络

2.2 产品核心特性

特性	说明
HTTP接口开放	适用于任何支持HTTP请求的编程语言，支持Web、APP/小程序、SaaS/低代码平台
状态实时上报	设备状态变化时主动推送消息，平台支持HTTP/MQTT两种方式接收
免网关直连	WiFi直接连接，无需网关支持
大功率承载	30A/6600W，覆盖工业设备、充电桩等场景
私有化部署	支持自建消息服务器，满足数据安全合规要求
离线执行	支持定时任务、点动控制等，断网仍可执行预设任务

2.3 选型

场景	推荐型号	说明
工业设备	智能断路器30A	三相/大功率单相设备
充电桩	智能断路器30A	交流充电桩标准配置
智能家居总闸	智能断路器30A	家庭入户总开关
多路设备	智能通断器AC4-30A	4路独立控制，每路30A

三、系统集成设计

3.1 两种状态获取方式对比

芯步平台提供两种方式获取设备状态，可根据项目需求选择：

方式	原理	适用场景
消息推送（推荐）	设备状态变化时平台主动推送	实时监控、告警联动、状态同步
主动查询	自行调用API查询设备状态	低频检查、简单场景

消息推送是本方案的核心推荐方式，无需频繁轮询，状态变化即时知晓，延迟低、效率高。

3.2 架构全景图（消息推送模式）

┌─────────────────────────────────────────────────────────────────┐
│                      应用层（业务系统）                          │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐          │
│  │ 管理后台 │ │ 监控大屏 │ │ 告警系统 │ │ 运维APP  │          │
│  └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘          │
└───────┼────────────┼────────────┼────────────┼─────────────────┘
        │            │            │            │
        ▼            ▼            ▼            ▼
┌─────────────────────────────────────────────────────────────────┐
│                    业务服务器（接收推送）                        │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │  状态存储 │ 事件处理 │ 告警判断 │ 业务逻辑             │    │
│  └─────────────────────────────────────────────────────────┘    │
└───────────────────────────────▲─────────────────────────────────┘
                                │ HTTPS / MQTT
                                │ （状态变化主动推送）
┌─────────────────────────────────────────────────────────────────┐
│                      芯步云平台（消息推送）                  │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │  状态采集 │ 消息路由 │ 签名验证 │ 推送服务             │    │
│  └─────────────────────────────────────────────────────────┘    │
└───────────────────────────────▲─────────────────────────────────┘
                                │ 设备上报
                                │
┌─────────────────────────────────────────────────────────────────┐
│                      执行层（30A智能开关）                       │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │          30A断路器（实时上报通断状态）                   │    │
│  └─────────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────────┘

3.3 数据流说明

数据类型	方向	说明
控制指令	业务系统 → 平台 → 设备	下发开/关命令
状态上报	设备 → 平台 → 业务系统	设备状态变化自动推送
心跳监测	设备 → 平台	定期上报在线状态

四、设备状态查询实现方案

4.1 方式一：消息推送接收状态（推荐）

这是获取设备状态的推荐方式，设备状态变化时平台会主动推送消息到你的服务器，无需轮询查询。

4.1.1 配置消息推送

登录芯步物联网控制台，开启消息推送功能：

进入「物联网控制台」→「消息推送」
设置接收消息的服务器URL（HTTP方式）或MQTT配置
保存配置，平台将自动推送设备状态变化

两种接收方式可选：

接收方式	说明	推荐场景
HTTP方式	平台推送消息到指定URL	简单集成，无需额外配置
MQTT方式	订阅 `api/{AppId}/message/#` 主题接收消息	低延迟场景，推荐使用

4.1.2 推送消息格式

当设备状态变化时，平台推送的HTTP消息格式如下

关键字段说明

type: "state" 表示这是一条状态上报消息
data[].power 为当前开关状态（字符串类型）
ts 为设备上报时间

4.1.3 Python接收示例（HTTP方式）

from flask import Flask, request, jsonify
import json

app = Flask(__name__)

@app.route('/device/state/callback', methods=['POST'])
def device_state_callback():
    """接收芯步平台推送的设备状态"""
    try:
        # 获取推送的消息
        data = request.get_json()
        print(f"收到消息: {json.dumps(data, indent=2)}")
        
        # 解析状态数据
        device_id = data.get('device')
        msg_type = data.get('type')
        
        if msg_type == 'state':
            message = data.get('message', {})
            data_list = message.get('data', [])
            
            for item in data_list:
                if 'power' in item:
                    power_state = item['power']  # "1" 或 "0"
                    timestamp = message.get('ts')
                    
                    # 处理状态更新
                    update_device_state(device_id, power_state, timestamp)
                    
                    # 检查是否需要告警
                    if power_state == '0':
                        check_and_alert(device_id, '开关已断开')
        
        return jsonify({"code": 0, "message": "success"}), 200
        
    except Exception as e:
        print(f"处理异常: {e}")
        return jsonify({"code": -1, "message": str(e)}), 200  # 注意:5秒超时

def update_device_state(device_id, state, timestamp):
    """更新数据库中的设备状态"""
    # 实际项目中应写入数据库
    print(f"[DB] 设备{device_id}状态更新为: {'开启' if state == '1' else '关闭'}，时间: {timestamp}")

def check_and_alert(device_id, message):
    """告警判断与发送"""
    # 可根据业务规则发送告警
    print(f"[ALERT] 设备{device_id}: {message}")

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

4.1.4 Python接收示例（MQTT方式）

import paho.mqtt.client as mqtt
import json

APP_ID = "your_app_id"

def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code: {rc}")
    # 订阅状态主题
    client.subscribe(f"api/{APP_ID}/message/state")
    # 也可订阅所有消息
    # client.subscribe(f"api/{APP_ID}/message/#")

def on_message(client, userdata, msg):
    """处理收到的MQTT消息"""
    try:
        payload = json.loads(msg.payload.decode())
        print(f"Topic: {msg.topic}")
        print(f"Message: {json.dumps(payload, indent=2)}")
        
        # 解析状态
        if payload.get('type') == 'state':
            device_id = payload.get('device')
            message = payload.get('message', {})
            data_list = message.get('data', [])
            
            for item in data_list:
                if 'power' in item:
                    power_state = item['power']
                    print(f"设备{device_id}状态: {'开启' if power_state == '1' else '关闭'}")
                    
    except Exception as e:
        print(f"解析异常: {e}")

# 配置MQTT客户端
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

# 连接MQTT Broker（配置地址和端口）
client.connect("mqtt.thingboot.com", 1883, 60)
client.loop_forever()

4.2 方式二：主动查询设备状态

如果无法使用消息推送，也可通过API主动查询设备状态。

4.2.1 认证签名算法

芯步采用双重MD5签名机制确保API调用安全

签名计算流程:
1. step1 = md5(AppSecret)        // 对开发者密码进行一次MD5
2. step2 = step1 + ts             // 拼接当前Unix时间戳（秒）
3. sign = md5(step2)              // 对拼接结果再次MD5

参数说明

AppID：应用ID，在控制台获取
AppSecret：开发者密码
ts：当前Unix时间戳（秒）

4.2.2 查询设备状态API

import hashlib
import time
import requests

class Breaker30A:
    """30A智能断路器控制器"""
    
    def __init__(self, app_id: str, app_secret: str):
        self.app_id = app_id
        self.app_secret = app_secret
        self.base_url = f"https://api.thingboot.com/{app_id}/device/control/"
    
    def _generate_sign(self, ts: int) -> str:
        """生成API签名（双重MD5）"""
        step1 = hashlib.md5(self.app_secret.encode()).hexdigest()
        step2 = step1 + str(ts)
        sign = hashlib.md5(step2.encode()).hexdigest()
        return sign
    
    def get_status(self, device_id: str) -> dict:
        """主动查询设备开关状态"""
        ts = int(time.time())
        sign = self._generate_sign(ts)
        
        url = f"{self.base_url}?sign={sign}&ts={ts}"
        
        # 查询设备状态的order格式
        payload = {
            "device": device_id,
            "order": {"get_status": 1}  # 获取状态命令
        }
        
        response = requests.post(url, json=payload, timeout=10)
        return response.json()
    
    def control(self, device_id: str, state: int) -> dict:
        """控制开关"""
        ts = int(time.time())
        sign = self._generate_sign(ts)
        
        url = f"{self.base_url}?sign={sign}&ts={ts}"
        payload = {
            "device": device_id,
            "order": {"power": state}
        }
        
        response = requests.post(url, json=payload, timeout=10)
        return response.json()

# 使用示例
breaker = Breaker30A("your_app_id", "your_app_secret")

# 查询状态
status = breaker.get_status("device_001")
print("设备状态:", status)

# 控制开关
result = breaker.control("device_001", 1)  # 开启
print("控制结果:", result)

五、完整对接实现代码

5.1 Python完整集成示例

以下是一个完整的30A智能开关集成类，支持控制命令下发和状态消息接收：

import hashlib
import time
import requests
import json
from typing import Optional

class SmartBreaker30A:
    """30A智能断路器完整控制类"""
    
    def __init__(self, app_id: str, app_secret: str, device_id: str):
        self.app_id = app_id
        self.app_secret = app_secret
        self.device_id = device_id
        self.base_url = f"https://api.thingboot.com/{app_id}/device/control/"
        self._current_state = None  # 缓存当前状态
    
    def _generate_sign(self, ts: int) -> str:
        """生成API签名（双重MD5）"""
        step1 = hashlib.md5(self.app_secret.encode()).hexdigest()
        step2 = step1 + str(ts)
        return hashlib.md5(step2.encode()).hexdigest()
    
    def _send_command(self, order: dict) -> dict:
        """发送命令到设备"""
        ts = int(time.time())
        sign = self._generate_sign(ts)
        
        url = f"{self.base_url}?sign={sign}&ts={ts}"
        payload = {"device": self.device_id, "order": order}
        
        response = requests.post(url, json=payload, timeout=10)
        return response.json()
    
    # ========== 控制命令 ==========
    
    def turn_on(self) -> dict:
        """开启开关"""
        return self._send_command({"power": 1})
    
    def turn_off(self) -> dict:
        """关闭开关"""
        return self._send_command({"power": 0})
    
    def pulse_control(self, duration_ms: int = 500) -> dict:
        """点动控制:通电duration_ms毫秒后自动断电"""
        return self._send_command({"point": str(duration_ms)})
    
    def reboot(self, duration_ms: int = 10000) -> dict:
        """重启:断电后自动恢复"""
        return self._send_command({"reset": str(duration_ms)})
    
    # ========== 状态管理 ==========
    
    def update_state(self, state: str):
        """更新本地缓存状态（由推送回调调用）"""
        self._current_state = state
    
    def get_cache_state(self) -> Optional[str]:
        """获取缓存的状态"""
        return self._current_state
    
    def is_on(self) -> bool:
        """判断是否开启"""
        return self._current_state == "1"
    
    def is_off(self) -> bool:
        """判断是否关闭"""
        return self._current_state == "0"

# ========== 业务集成示例 ==========

class EnergyMonitor:
    """能源监控系统"""
    
    def __init__(self):
        self.devices = {}  # device_id -> SmartBreaker30A实例
    
    def register_device(self, device_id: str, app_id: str, app_secret: str):
        """注册设备"""
        self.devices[device_id] = SmartBreaker30A(app_id, app_secret, device_id)
    
    def on_state_push(self, data: dict):
        """处理平台推送的状态消息"""
        device_id = data.get('device')
        message = data.get('message', {})
        data_list = message.get('data', [])
        
        for item in data_list:
            if 'power' in item:
                state = item['power']
                
                # 更新本地缓存
                if device_id in self.devices:
                    self.devices[device_id].update_state(state)
                
                # 记录状态变化
                self.log_state_change(device_id, state)
                
                # 触发告警检查
                self.check_alerts(device_id, state)
    
    def log_state_change(self, device_id: str, state: str):
        """记录状态变化到数据库"""
        print(f"[LOG] {device_id} 状态变更为: {'开启' if state == '1' else '关闭'}")
        # 实际项目中写入数据库
    
    def check_alerts(self, device_id: str, state: str):
        """检查是否需要告警"""
        if state == "0":
            # 开关断开，检查是否异常
            print(f"[INFO] 设备{device_id}已断开")
    
    def remote_control(self, device_id: str, action: str) -> dict:
        """远程控制"""
        if device_id not in self.devices:
            return {"error": "设备不存在"}
        
        device = self.devices[device_id]
        if action == "on":
            return device.turn_on()
        elif action == "off":
            return device.turn_off()
        elif action == "reboot":
            return device.reboot()
        
        return {"error": "未知操作"}

# ========== 使用示例 ==========
if __name__ == "__main__":
    # 初始化监控系统
    monitor = EnergyMonitor()
    monitor.register_device(
        device_id="breaker_001",
        app_id="your_app_id",
        app_secret="your_app_secret"
    )
    
    # 场景1:推送消息处理
    push_data = {
        "device": "breaker_001",
        "type": "state",
        "message": {
            "mid": "abc123",
            "data": [{"power": "1"}],
            "ts": "1656416279506"
        }
    }
    monitor.on_state_push(push_data)
    
    # 场景2:远程控制
    result = monitor.remote_control("breaker_001", "on")
    print("控制结果:", result)
    
    # 场景3:直接操作设备
    breaker = SmartBreaker30A("your_app_id", "your_app_secret", "breaker_001")
    print("开启开关:", breaker.turn_on())
    print("当前状态:", breaker.get_cache_state())

5.2 cURL命令行示例

查询设备状态

六、典型应用场景配置

6.1 第一种场景：工业设备远程监控

需求：实时监控重要设备开关状态，异常断电时第一时间告警。

实现流程

设备状态变化 → 平台推送 → 业务系统接收 → 更新数据库 → 判断告警 → 发送通知

代码实现

6.2 第二种场景：充电桩状态同步

需求：充电桩运营平台需要实时同步每个充电桩的通断电状态。

实现方案

每台充电桩配置30A智能断路器
平台通过消息推送接收状态变化
用户端展示实时状态

6.3 第三种场景：远程运维与故障诊断

需求：设备死机时，远程断电重启并确认恢复状态。

七、实施部署指南

7.1 硬件安装

步骤	说明
1. 断电操作	安装前切断上游电源
2. 导轨安装	标准35mm DIN导轨安装
3. 接线连接	严格按照产品手册接线，输入端接电源，输出端接负载
4. 配网配置	通过控制台或小程序配置WiFi
5. 功能测试	测试远程控制和状态上报

7.2 配网指南

芯步提供两种配网方式

方式一：电脑控制台配网

登录芯步官网物联网控制台
点击「网络配置」→「登记网络」，输入WiFi名称和密码
点击已添加的网络，选择「为设备配置此网络」
按照页面提示用手机热点配网

方式二：小程序配网

微信搜索「芯步小程序」并登录
选择工作台 → 点击「+」添加设备
按提示用手机热点为设备配网

7.3 网络要求

WiFi频段：仅支持2.4GHz
信号强度：≥-70dBm
网络备份：支持5组WiFi配置，自动切换
私有化部署：支持自建消息服务器，纯局域网内运行

7.4 消息推送配置要点

配置项	说明
推送URL	必须是公网可访问地址（或内网可访问，如私有化部署）
超时时间	5秒内需返回HTTP 200状态码，否则推送失败
日志查看	可在控制台查看推送日志排查问题

八、总结

对比维度	传统方案	芯步30A智能开关方案
状态获取	需现场查看或人工上报	主动推送，实时感知
控制方式	手动操作	HTTP API远程控制
告警时效	事后发现，滞后严重	状态变化即时推送，秒级响应
系统集成	无开放接口	标准HTTP API，1-2人日完成
承载能力	普通开关功率有限	30A/6600W，可覆盖工业设备
网络依赖	无需网络	WiFi连接，支持私有化部署

集成工作量评估

消息推送接收配置：0.5-1人日
控制API对接：1人日
完整业务系统集成（含告警、日志）：2-3人日

通过芯步30A智能断路器，可实现大功率设备远程控制、实时状态感知、异常主动告警的智能化配电管理闭环，显著提升运维效率和设备安全性。