怎样接入8 路包间电器覆盖控制器以实现包间设备故障自动告警_解决方案

CATALOG

一、概述

共享包间（棋牌室、茶室、影音房、舞蹈室等）内电器设备众多——空调、照明、麻将机、投影仪、排风扇等。设备故障是运营中的高频痛点：一台空调不制冷无人发现、一盏灯坏掉影响用户体验、排风扇异响未被察觉……传统方案依赖人工巡检，响应慢、遗漏多。

芯步8路包间电器覆盖控制器通过开放HTTP API + 消息推送机制，不仅提供8路独立电源远程控制，更能实时感知设备状态变化，实现设备故障自动告警。您的服务器接收设备上报的状态数据后，通过自定义规则引擎判断故障并触发告警，无需现场巡检即可掌握包间设备健康状态。

核心能力

8路独立分控：单台设备覆盖包间内所有电器（照明、空调、麻将机、排风扇、门禁）
实时状态上报：设备通电/断电状态变化时，毫秒级推送至您的服务器
开放HTTP API：适用于Web、APP/小程序、SaaS等任意软件项目
私有化部署：支持局域网运行，数据安全可控
即装即用：WiFi直连无需网关，可设定5组WiFi自动切换

典型应用场景

场景	设备类型	告警逻辑
空调故障监测	空调专用回路	用户开启空调后，5分钟内功率＜阈值→推送"空调可能无制冷"告警
照明异常告警	照明回路	某路照明通电但电压异常→推送"照明故障，请检查灯泡"
排风扇卡死/异响	换气扇回路	通电后电流/功率波动异常→推送"排风扇可能卡死"
设备离线通知	控制器本体	设备连续5分钟不在线→推送"包间控制器离线，请检查网络"
门禁状态反馈	门禁回路	供电正常但锁未吸合→推送"电磁锁可能故障"

二、核心产品选型

2.1 8路包间电器覆盖控制器

芯步提供多种8路控制器，可根据包间需求选择：

型号	接口配置	适用场景
智能包间控制器\|Max	3路10A开关+3路16A插座+1路门禁+1路30A空调	完整包间（8路全功能）
智能控制器8路\|交流版	8路交流10A输出	照明、排风扇、普通电器
智能通用控制器\|8路	8路直流/交流负载	定制场景

推荐选择：共享包间场景推荐使用智能包间控制器|Max，专为包间设计，接口配置完整。

2.2 智能包间控制器|Max 详细参数

特性	参数
产品型号	UNI-KZQ-BJ-Max
控制路数	8路独立分控
回路1-3	10A开关接口 → 照明/排风扇/氛围灯（阻性≤2200W，感性≤350W）
回路4-6	16A插座接口 → 麻将机/饮水机/按摩椅（阻性≤3500W，感性≤500W）
回路7	10A门禁接口 → 电磁锁/电插锁（直流，直接控制）
回路8	30A插座接口 → 2匹空调专用（阻性≤6600W，感性≤1100W）
工作电压	DC 5V 2A
无线连接	WiFi 2.4GHz
控制接口	HTTP API
语音播报	可选TTS版本
私有化部署	支持局域网直连

数据来源：

2.3 产品核心特性

8路独立分控：每路均可通过HTTP接口远程独立控制通断
支持远程定时任务：平台内置定时功能，实现自动化管理
开放HTTP接口：适用于任何支持HTTP请求的编程语言
私有化部署：支持自建消息服务器，可运行在纯局域网环境
无网关设计：WiFi直连，可设定5组WiFi自动切换
响应延迟：从命令下发到设备响应约80-120ms

数据来源：

三、故障告警设计

3.1 整体架构图

┌─────────────────────────────────────────────────────────────────┐
│                    包间管理系统 / 运维平台                        │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐               │
│  │  规则引擎   │  │  告警中心   │  │  告警推送   │               │
│  │ （故障判断） │  │ （告看板）  │  │（钉钉/企微） │               │
│  └─────────────┘  └─────────────┘  └─────────────┘               │
└─────────────────────────────────────────────────────────────────┘
                              ▲
                              │ 消息推送（状态变化实时推送）
                              │ 命令下发（重启/断电等补救操作）
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                     芯步云平台 / 私有化服务器                  │
│                    （消息推送 + 命令路由）                         │
└─────────────────────────────────────────────────────────────────┘
                              ▲
                              │ WiFi 2.4GHz
                              │ 状态上报（每秒/状态变化时）
                              │ 命令下达
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                  8路包间控制器（Max）                             │
│  ┌──────┬──────┬──────┬──────┬──────┬──────┬──────┬──────┐       │
│  │回路1 │回路2 │回路3 │回路4 │回路5 │回路6 │回路7 │回路8 │       │
│  │照明 │排风 │氛围 │麻将机│茶具 │按摩椅│门禁 │空调  │       │
│  └──────┴──────┴──────┴──────┴──────┴──────┴──────┴──────┘       │
│                            ▲              ▲                      │
│                            │              │                      │
│                    通电/断电状态    电压/电流/功率数据             │
│                     变化即上报      （需搭配计量功能）             │
└─────────────────────────────────────────────────────────────────┘

3.2 两种状态获取方式

方式	实现方法	延迟	适用场景
消息推送（推荐）	设备状态变化时平台主动推送至您的服务器	毫秒级	实时告警、联动响应
主动查询	通过API定时轮询设备状态	秒级-分钟级	非实时监控、数据统计

消息推送是告警的核心：设备每次状态变化（通电/断电/故障）都会实时推送，您的服务器接收后即可触发规则判断。

3.3 故障告警数据流

1. 用户开启空调（系统下发 `power8: "1"`）
                ↓
2. 空调回路通电，设备状态变化
                ↓
3. 平台向您的服务器推送状态消息
   {"device":"room_101","type":"state","message":{"data":[{"power8":"1"}]}}
                ↓
4. 您的规则引擎开始计时（如设定5分钟监测窗口）
                ↓
5. 等待空调实际运行数据反馈
   - 如有计量版:监测功率值是否达标
   - 如无计量版:通过温湿度传感器间接判断
                ↓
6. 规则判断:
   - 功率＞阈值 → 正常
   - 功率＜阈值持续N分钟 → 触发"空调不制冷"告警
                ↓
7. 告警中心推送通知（钉钉/企业微信/短信）

四、API接口接入详解

4.1 接口地址与签名计算

请求地址

https://api.thingboot.com/{AppId}/device/control/?sign={sign}&ts={ts}

签名计算公式

sign = md5( md5(AppSecret) + ts )

其中ts为当前Unix时间戳（秒），AppSecret为开发者密码。

签名计算示例

签名方法参考：

4.2 核心控制命令

4.2.1 单路独立控制

4.2.2 批量控制（场景切换）

命令格式参考：

4.2.3 先断后通（设备重启）

应用：空调主机死机时，断电10秒后重启

4.2.4 携带业务标识

extra字段会在消息推送中原样返回，用于关联业务订单

五、消息推送接入（故障告警核心）

5.1 配置消息推送

登录芯步控制台
进入「开发设置」→「消息推送」
配置您的服务器接收URL（如 https://your-domain.com/api/device/callback）
选择需要接收的消息类型（设备类设备选中state状态消息）

配置方法参考：

5.2 状态上报数据格式

当设备状态变化时（如某路通电/断电），平台向您的服务器推送：

数据格式参考：

六、故障告警规则引擎实现

6.1 核心实现代码

from flask import Flask, request, jsonify
import requests
import time
import hashlib
import threading
import schedule
from datetime import datetime

app = Flask(__name__)

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

# 设备配置:包间ID -> {控制器ID, 回路配置, 告警规则}
ROOMS = {
    "room_101": {
        "device_id": "controller_101",
        "relays": {
            1: {"name": "照明", "type": "light", "alarm": {"enabled": True}},
            2: {"name": "排风扇", "type": "fan", "alarm": {"enabled": True, "check_duration": 120}},
            8: {"name": "空调", "type": "ac", "alarm": {"enabled": True, "threshold_power": 500, "check_duration": 300}},
        }
    }
}

# 存储待监测的设备状态
monitoring_tasks = {}

def generate_sign(ts):
    """生成API签名"""
    md5_secret = hashlib.md5(APP_SECRET.encode()).hexdigest()
    sign_str = md5_secret + str(ts)
    return hashlib.md5(sign_str.encode()).hexdigest()

def control_device(device_id, order):
    """向设备下发控制命令"""
    ts = int(time.time())
    sign = generate_sign(ts)
    url = f"https://api.thingboot.com/{APP_ID}/device/control/?sign={sign}&ts={ts}"
    payload = {"device": device_id, "order": order}
    
    try:
        response = requests.post(url, json=payload, timeout=5)
        return response.json()
    except Exception as e:
        print(f"控制失败: {e}")
        return None

def send_alert(room_id, device_name, alarm_type, message):
    """发送告警通知（集成钉钉/企微/短信）"""
    alert_msg = f"【{room_id}告警】{device_name} {message}"
    print(f"发送告警: {alert_msg}")
    
    # 示例:发送到钉钉机器人
    # requests.post("https://oapi.dingtalk.com/robot/send?access_token=xxx", 
    #                json={"msgtype": "text", "text": {"content": alert_msg}})
    
    # 记录告警到数据库
    # save_alarm_log(room_id, device_name, alarm_type, alert_msg)

def check_device_power(room_id, relay_num, actual_power):
    """检查设备功率是否正常（用于计量版）"""
    config = ROOMS.get(room_id, {}).get("relays", {}).get(relay_num, {})
    alarm_config = config.get("alarm", {})
    
    if not alarm_config.get("enabled"):
        return
    
    threshold = alarm_config.get("threshold_power")
    if threshold and actual_power < threshold:
        send_alert(room_id, config["name"], "power_low", 
                   f"通电后功率{actual_power}W低于阈值{threshold}W，可能故障")

def schedule_device_check(room_id, relay_num, relay_name, check_seconds):
    """延时检查设备状态（无计量版时使用）"""
    def check():
        # 主动查询设备状态判断是否异常
        # 此处可调用设备状态查询接口
        send_alert(room_id, relay_name, "no_response", 
                   f"通电{check_seconds//60}分钟后仍未检测到正常工作反馈，可能故障")
    
    timer = threading.Timer(check_seconds, check)
    if room_id not in monitoring_tasks:
        monitoring_tasks[room_id] = {}
    monitoring_tasks[room_id][relay_num] = timer
    timer.start()

def cancel_device_check(room_id, relay_num):
    """取消设备的延时检查"""
    if room_id in monitoring_tasks and relay_num in monitoring_tasks[room_id]:
        monitoring_tasks[room_id][relay_num].cancel()
        del monitoring_tasks[room_id][relay_num]

@app.route('/api/device/callback', methods=['POST'])
def device_callback():
    """接收设备状态消息推送 - 告警核心入口"""
    try:
        data = request.get_json()
        print(f"收到设备消息: {data}")
        
        device_id = data.get('device')
        msg_data = data.get('message', {}).get('data', [{}])[0]
        
        # 识别是哪个包间的哪个回路状态变化
        for room_id, config in ROOMS.items():
            if config["device_id"] == device_id:
                for relay_num, relay_config in config["relays"].items():
                    power_key = f"power{relay_num}"
                    if power_key in msg_data:
                        power_state = msg_data[power_key]
                        
                        if power_state == "1":
                            # 设备通电:启动故障监测
                            print(f"{room_id} {relay_config['name']} 已通电，启动监测")
                            # 无计量版时用延时检查替代
                            schedule_device_check(
                                room_id, relay_num, relay_config["name"], 
                                relay_config.get("alarm", {}).get("check_duration", 300)
                            )
                        else:
                            # 设备断电:取消监测
                            cancel_device_check(room_id, relay_num)
                        break
                break
        
        return jsonify({"code": 200}), 200
        
    except Exception as e:
        print(f"处理消息异常: {e}")
        return jsonify({"code": 500}), 500

@app.route('/api/device/power_report', methods=['POST'])
def power_report():
    """接收计量版功率上报（若有）"""
    data = request.get_json()
    device_id = data.get('device')
    power_watts = data.get('power', 0)
    
    # 与之前通电状态对比，判断设备是否正常工作
    # check_device_power(room_id, relay_num, power_watts)
    
    return jsonify({"code": 200}), 200

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

6.2 告警规则配置示例

七、告警场景实战

7.1 第一种场景：空调不制冷自动告警

需求：用户开启空调后，若5分钟内空调未正常制冷，系统自动推送告警。

实现方案

监听空调回路（回路8）通电状态变化
通电后启动5分钟计时
配合温度传感器或计量模块判断制冷效果
无改善则触发告警并可选远程重启空调

7.2 第二种场景：设备离线告警

需求：包间控制器离线超过5分钟，系统自动通知运维人员。

7.3 第三种场景：批量控制失败告警

需求：批量下发指令时，若有设备未成功执行，自动告警。

八、私有化部署方案

共享包间运营方对数据安全要求较高，芯步8路控制器支持纯局域网运行。

部署模式对比

特性	公有云模式	私有化模式
设备联网	WiFi连接互联网	连接包间内网WiFi
API地址	`api.thingboot.com`	自建服务器IP
数据流向	经过芯步云端	内网直连，数据不出门店
网络依赖	需要互联网	仅需局域网

私有化配置要点

设备支持配置5组WiFi网络，自动连接信号最强的网络
需自建HTTP服务，所有API调用和消息推送在内网完成
适合连锁品牌统一部署

九、设备安装部署指南

9.1 安装要求

项目	要求
安装方式	标准配电箱/DIN导轨安装
工作电压	DC 5V/12V（视型号）
各回路负载	2200W-6600W（视回路类型）
网络要求	WiFi 2.4GHz信号覆盖良好

9.2 包间回路分配

回路	推荐用途	负载类型	状态监测要点
回路1	主照明	10A开关	通电后电压/电流是否正常
回路2	排风扇	10A开关	电流波动→判断是否卡死
回路3	氛围灯/吸烟灯	10A开关	通电后功率是否达标
回路4-6	麻将机/茶具/按摩椅	16A插座	通电后有无工作反馈
回路7	门禁电磁锁	10A门禁	供电后锁是否吸合
回路8	空调	30A插座	功率/温度双重验证

十、常见问题与排障

问题	可能原因	解决方案
设备不在线	WiFi信号弱	检查WiFi覆盖，设备支持5组WiFi自动切换
收不到消息推送	回调地址不可达	检查服务器是否可达，需5秒内返回200
误报空调故障	检测逻辑不合理	延长检测窗口，增加温度变化阈值
告警重复发送	未做去重处理	增加告警去重机制（同设备5分钟内只告一次）
命令下发成功但设备无响应	设备离线	200仅表示平台接收，需通过消息推送确认执行
签名验证失败	参数错误	确认ts为秒级时间戳，公式为`md5(md5(AppSecret)+ts)`

十一、总结

本方案基于芯步8路包间控制器，实现了包间电器的远程控制+故障自动告警闭环：

能力	实现的方式是
8路独立分控	每路独立API控制
实时状态感知	消息推送机制，状态变化毫秒级上报
自定义故障检测	规则引擎+延时检查架构
多告警渠道	钉钉/企微/短信集成
远程故障恢复	`reset`命令远程重启
私有化部署	内网运行，数据安全

集成核心步骤

根据包间需求选择8路控制器型号（推荐Max版）
获取AppID/AppSecret，实现MD5签名算法
配置消息推送URL，接收设备状态数据
开发规则引擎，实现故障判断和告警推送
物理安装设备，按回路分配连接电器

方案优势

✅ 8路全覆盖：单设备管理包间所有电器
✅ 接口统一：标准HTTP API，10分钟完成对接
✅ 实时感知：状态变化毫秒级推送，无轮询延迟
✅ 告警精准：规则引擎自定义，避免误报漏报
✅ 远程运维：80-120ms响应，故障可远程重启
✅ 私有化支持：数据安全可控，适合连锁品牌

如需详细的产品手册和API文档，请访问查阅或联系技术支持团队获取。