如何二次开发2路智能开关以实现线路状态反馈控制_解决方案

CATALOG

芯步的2路智能开关提供了完整的HTTP API接口，支持独立控制每一路继电器，同时可通过回调机制获取实时状态。以下是基于其开放接口的二次开发解决方案。

1. 概述与准备

1.1 适用产品

本方案适用于芯步以下2路智能硬件产品：

智能触摸墙壁开关2路：标准86型墙壁开关，适用于家居照明改造
智能控制器2路：直流电压版，适用于自动化设备控制

1.2 核心功能特性

双路独立控制：支持对线路1、线路2分别进行开/关操作
实时状态反馈：设备状态变化可主动推送至开发者服务器
多种控制模式：支持点动（先通后断）、互锁（先断后通）等工业控制模式
开放接口：提供HTTP API和MQTT两种对接方式

1.3 二次开发准备工作

项目	说明	获取方式
AppID	应用唯一标识	登录控制台 → 开发设置
AppSecret	开发者密钥（用于签名）	登录控制台 → 开发设置
设备ID	目标设备的唯一标识	控制台设备列表 / API拉取
消息接收URL	接收设备状态推送的服务器地址	开发者自建，在控制台配置

签名算法

Sign = MD5(MD5(AppSecret) + ts)

其中 ts 为当前Unix时间戳（秒），+ 表示字符串拼接

2. 接口调用详解

2.1 核心API地址

POST https://api.thingboot.com/{AppID}/device/control/?sign={sign}&ts={ts}
Content-Type: application/json

2.2 支持的设备命令

命令类型	命令格式	功能说明	适用场景
线路1开关	`{"power1":1}` 或 `{"power1":0}`	控制第1路通断	基础开关控制
线路2开关	`{"power2":1}` 或 `{"power2":0}`	控制第2路通断	基础开关控制
批量控制	`{"power1":1,"power2":0}`	同时控制两路	场景联动
状态保持	`{"power1":{"keep":"1","revert":"3"}}`	用户手动操作后自动恢复	酒店门牌、公共照明
先通后断	`{"point1":"2000"}`	先接通，延时后断开（单位毫秒）	门禁控制、脉冲触发
先断后通	`{"reset1":"2000"}`	先断开，延时后接通（单位毫秒）	设备重启、复位控制

2.3 签名计算与请求示例（Python）

import hashlib
import time
import requests
import json

# 配置参数
APP_ID = "your_app_id"
APP_SECRET = "your_app_secret"
DEVICE_ID = "your_device_id"

def generate_sign(secret):
    """生成签名:md5(md5(secret) + timestamp)"""
    ts = str(int(time.time()))
    md5_secret = hashlib.md5(secret.encode()).hexdigest()
    sign_str = md5_secret + ts
    sign = hashlib.md5(sign_str.encode()).hexdigest()
    return sign, ts

def control_device(power1=None, power2=None, point1=None, point2=None, reset1=None, reset2=None):
    """
    控制2路智能开关
    参数:power1/power2: 1=开，0=关
         point1/point2: 先通后断，单位毫秒
         reset1/reset2: 先断后通，单位毫秒
    """
    sign, ts = generate_sign(APP_SECRET)
    url = f"https://api.thingboot.com/{APP_ID}/device/control/?sign={sign}&ts={ts}"
    
    # 构建命令
    order = {}
    if power1 is not None: order["power1"] = power1
    if power2 is not None: order["power2"] = power2
    if point1 is not None: order["point1"] = point1
    if point2 is not None: order["point2"] = point2
    if reset1 is not None: order["reset1"] = reset1
    if reset2 is not None: order["reset2"] = reset2
    
    payload = {
        "device": DEVICE_ID,
        "order": order
    }
    
    response = requests.post(url, json=payload)
    return response.json()

# 示例调用
# 1. 打开第一路，关闭第二路
result = control_device(power1=1, power2=0)

# 2. 第二路执行点动（先通后断，持续2秒）
result = control_device(point2=2000)

2.4 MQTT方式（推荐实时性要求高的场景）

2.5 分组控制（批量控制多设备）

当需要同时控制多个2路开关时，可使用分组控制接口

POST https://api.thingboot.com/{AppID}/group/control/?sign={sign}&ts={ts}

3. 线路状态反馈实现

3.1 实时状态反馈机制

芯步平台支持将设备状态变化主动推送到开发者指定的URL，实现实时反馈。

配置方式

登录芯步控制台
进入"开发设置" → "消息推送"
设置接收状态推送的URL（如 https://your-domain.com/api/device/callback）

推送的数据格式

3.2 HTTP回调接收服务实现

3.3 主动查询设备状态（可选方案）

若不配置消息推送，也可通过主动查询方式获取状态：

4. 完整二次开发架构

4.1 系统架构图

┌─────────────────────────────────────────────────────────────┐
│                      业务应用层                              │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────────┐       │
│  │ Web应用 │ │ 移动APP │ │ 小程序  │ │ 桌面客户端  │       │
│  └────┬────┘ └────┬────┘ └────┬────┘ └──────┬──────┘       │
│       └───────────┴───────────┴─────────────┘               │
│                         │ HTTP/HTTPS                        │
│  ┌──────────────────────▼──────────────────────┐           │
│  │           开发者业务服务器                    │           │
│  │  ┌─────────────┐  ┌─────────────────────┐  │           │
│  │  │ 设备状态管理 │  │ 业务逻辑处理        │  │           │
│  │  │ 数据库      │  │ 用户权限控制        │  │           │
│  │  └─────────────┘  └─────────────────────┘  │           │
│  └──────┬───────────────────────────┬─────────┘           │
│         │ HTTP API调用               │ 状态推送接收        │
└─────────┼───────────────────────────┼─────────────────────┘
          │                           │
          ▼                           ▲
┌─────────────────────────────────────────────────────────────┐
│                   芯步云平台                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  API网关  │  设备管理  │  消息中心  │  数据存储     │   │
│  └─────────────────────────────────────────────────────┘   │
│                           │ MQTT/HTTP                       │
└───────────────────────────┼─────────────────────────────────┘
                            │
                            ▼
          ┌─────────────────────────────────┐
          │        2路智能开关设备           │
          │  ┌──────────┐  ┌──────────┐    │
          │  │ 线路1    │  │ 线路2    │    │
          │  │ 继电器   │  │ 继电器   │    │
          │  └────┬─────┘  └────┬─────┘    │
          └───────┼─────────────┼──────────┘
                  │             │
                  ▼             ▼
            电器设备A       电器设备B

4.2 完整业务流程

class TwoChannelSwitchController:
    """2路智能开关控制器封装类"""
    
    def __init__(self, app_id, app_secret, device_id, callback_url=None):
        self.app_id = app_id
        self.app_secret = app_secret
        self.device_id = device_id
        self.callback_url = callback_url
        self.status_cache = {"power1": None, "power2": None}
    
    def _generate_sign(self):
        """生成签名"""
        ts = str(int(time.time()))
        md5_secret = hashlib.md5(self.app_secret.encode()).hexdigest()
        sign = hashlib.md5((md5_secret + ts).encode()).hexdigest()
        return sign, ts
    
    def _request(self, order):
        """发送请求"""
        sign, ts = self._generate_sign()
        url = f"https://api.thingboot.com/{self.app_id}/device/control/?sign={sign}&ts={ts}"
        payload = {"device": self.device_id, "order": order}
        return requests.post(url, json=payload).json()
    
    def turn_on_channel1(self):
        """开启线路1"""
        return self._request({"power1": 1})
    
    def turn_off_channel1(self):
        """关闭线路1"""
        return self._request({"power1": 0})
    
    def turn_on_channel2(self):
        """开启线路2"""
        return self._request({"power2": 1})
    
    def turn_off_channel2(self):
        """关闭线路2"""
        return self._request({"power2": 0})
    
    def set_both(self, state1, state2):
        """同时设置两路状态"""
        return self._request({"power1": state1, "power2": state2})
    
    def pulse_channel1(self, duration_ms):
        """线路1点动（先通后断）"""
        return self._request({"point1": duration_ms})
    
    def reset_channel1(self, duration_ms):
        """线路1复位（先断后通）"""
        return self._request({"reset1": duration_ms})
    
    def set_keep_mode(self, channel, keep_state, revert_seconds):
        """
        设置状态保持模式
        :param channel: 1或2
        :param keep_state: 保持的状态，1=开，0=关
        :param revert_seconds: 用户操作后恢复的时间（秒）
        """
        key = f"power{channel}"
        return self._request({key: {"keep": str(keep_state), "revert": str(revert_seconds)}})
    
    def update_status(self, status_data):
        """更新本地状态缓存"""
        self.status_cache["power1"] = status_data.get("power1")
        self.status_cache["power2"] = status_data.get("power2")
        return self.status_cache
    
    def get_status(self):
        """获取当前缓存的状态"""
        return self.status_cache

5. 典型应用场景

5.1 智能家居：双路照明控制

5.2 工业自动化：电机正反转控制

5.3 共享设备：扫码控制

5.4 状态监控告警

6. 注意事项与应用

6.1 签名时效性

签名中的时间戳 ts 有效期为5分钟
每次请求重新生成，避免时间偏差导致认证失败

6.2 状态同步最佳实践

优先使用消息推送机制：避免轮询造成的API压力和延迟
维护本地状态缓存：结合推送更新和主动查询双重保障
配置重试机制：网络波动时自动重试，保证指令送达

6.3 安全

IP白名单：在控制台配置API调用的白名单IP
HTTPS协议：生产环境使用HTTPS保证通信加密
签名密钥：AppSecret存储在服务端，严禁暴露在客户端代码中

6.4 典型问题与解决

问题	可能原因	解决方案
401签名错误	时间戳偏差、密钥错误	检查服务器时间同步，重新核对AppSecret
设备无响应	设备离线或ID错误	检查设备网络状态，确认设备ID正确
状态推送未收到	回调URL不可达	检查URL外网可访问性，查看控制台推送日志
命令执行延迟	网络延迟或MQTT拥塞	切换至MQTT方式，或增加超时时间

6.5 扩展能力

定时任务：通过平台接口设置开关定时策略
场景联动：结合传感器设备（如人体感应、门磁）自动触发开关动作
数据分析：获取设备操作日志，分析使用模式和能耗情况

通过上述方案，开发者可以基于芯步2路智能开关的开放接口，快速实现双线路的独立控制与状态反馈，满足从智能家居到工业自动化的多样化应用需求。