共享麻将馆设备控制：怎么将16A可定时智能插座对接到项目中_解决方案

CATALOG

[16920] 文档版本：V1.0 更新日期：2026年5月 适用对象：软件工程师、共享棋牌室集成商、无人值守场馆开发者

第一章：共享麻将馆场景需求与产品选型

1.1 共享麻将馆的智能化需求

共享麻将馆作为无人值守经营场景，设备的远程控制和定时管理是运营的核心。麻将馆内主要设备及控制需求如下：

设备类型	典型功率	控制需求	推荐方案
麻将机	100-300W	订单开始通电，结束断电	16A智能插座
饮水机	500-800W	定时开关，节能管理	16A智能插座
空调	1000-2000W	预约开启，定时关闭	16A智能插座
照明	50-200W	人来灯亮，人走灯灭	传感器联动
排风扇	30-50W	定时通风	智能插座

1.2 芯步16A智能插座核心参数

针对麻将馆场景，芯步提供两款16A智能插座产品：

参数项	智能墙壁插座16A	智能插座16A
产品型号	UNI-QC-16A（墙壁型）	UNI-CZ-16A（即插型）
安装方式	86型底盒安装，替换原有插座	3pin插头，即插即用
适用场景	装修时预埋/线路改造	免安装，直接插在现有插座上
额定电流	MAX 16A	MAX 16A
阻性负载功率	MAX 3500W	MAX 3500W
感性负载功率	MAX 500W	MAX 500W
待机功耗	0.4W（断开）- 1W（接通）	0.4W（断开）- 1W（接通）
无线连接	WiFi 2.4GHz	WiFi 2.4GHz
外壳材质	防火V0级PC	防火V0级PC
功率计量	可选计量版	可选计量版

版本说明

不带功率计量版（UNI-QC-16A / UNI-CZ-16A）：基础远程控制功能
带功率计量版（UNI-QC-16A-P / UNI-CZ-16A-P）：额外支持电压、电流、功率实时监测，适合能耗统计场景

保姆级提示：若麻将机由LED灯带装饰，属感性负载，需控制在500W以内；墙壁型插座需86型底盒安装，即插型插座可免布线直接使用。

1.3 同一品类其他型号参考

芯步智能插座系列提供多个规格

型号	额定电流	负载功率	安装方式	适用场景
智能墙壁插座10A	MAX 10A	2200W	86型底盒	照明、风扇
智能墙壁插座16A	MAX 16A	3500W	86型底盒	空调、麻将机
智能插座10A	MAX 10A	2200W	即插即用	桌面电器
智能插座16A	MAX 16A	3500W	即插即用	大功率电器

第二章：HTTP接口协议详解

2.1 整体架构

芯步智能插座采用标准的HTTP请求-响应模型，适用于任何支持HTTP请求的编程语言

性能指标：从命令下发到设备实际响应，端到端延迟约为80-120ms。

2.2 请求地址格式

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

URL参数说明

参数	说明	示例
`{AppId}`	应用ID，在芯步控制台获取	`10001`
`sign`	动态签名	`a1b2c3d4...`
`ts`	当前Unix时间戳（秒级）	`1746000000`

2.3 签名算法（核心安全机制）

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

Sign = md5( md5(AppSecret) + ts )

签名计算步骤

Python示例

安全要点

两次MD5输出均为32位小写十六进制字符串
ts需与服务端时间同步，误差不超过5分钟
AppSecret需妥善保管，切勿硬编码在前端代码中

2.4 核心命令集

芯步16A智能插座支持以下核心命令

命令类型	命令示例	说明	麻将馆应用
通电	`{"power":"1"}`	插座通电	订单开始，麻将机启动
断电	`{"power":"0"}`	插座断电	订单结束，设备关闭
延时断电	`{"point":"3600000"}`	通电1小时后自动断电	定时关机
延时通电	`{"reset":"10000"}`	断电10秒后自动通电	设备重启
状态查询	`{"get_status":""}`	查询通断状态	订单开始前确认设备状态
计量查询	`{"metering":""}`	查询功率/电压/电流	能耗统计、异常监测

✨ 核心亮点：point 命令：支持“先通后断”——通电后由设备硬件定时断电，无需软件定时任务或后端队列。即使服务器宕机，插座也会在指定时间后自动断电，是共享场景的关键特性。

2.5 计量数据查询（带计量版）

命令格式

返回示例

2.6 完整请求示例

第三章：多语言代码实现

3.1 Python完整实现

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

# ========== 配置信息 ==========
APP_ID = "YOUR_APP_ID"           # 应用ID
APP_SECRET = "YOUR_APP_SECRET"   # 应用密钥
DEVICE_ID = "YOUR_DEVICE_ID"     # 插座设备ID
API_BASE = "https://api.thingboot.com"
# =============================

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

def send_command(order: dict, device_id: str = None) -> dict:
    """向设备发送命令"""
    ts = int(time.time())
    first_md5 = calculate_md5(APP_SECRET)
    sign_raw = first_md5 + str(ts)
    sign = calculate_md5(sign_raw)
    
    url = f"{API_BASE}/{APP_ID}/device/control/"
    params = {"sign": sign, "ts": ts}
    payload = {"device": device_id or 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 power_on(device_id: str = None) -> dict:
    """通电（开启设备）"""
    return send_command({"power": "1"}, device_id)

def power_off(device_id: str = None) -> dict:
    """断电（关闭设备）"""
    return send_command({"power": "0"}, device_id)

def delay_off(delay_ms: int, device_id: str = None) -> dict:
    """
    延时断电:通电后指定毫秒自动断电
    核心命令！无需软件定时任务
    """
    return send_command({"point": str(delay_ms)}, device_id)

def delay_on(delay_ms: int, device_id: str = None) -> dict:
    """延时通电:断电后指定毫秒自动通电（重启功能）"""
    return send_command({"reset": str(delay_ms)}, device_id)

def get_status(device_id: str = None) -> dict:
    """查询设备状态"""
    result = send_command({"get_status": ""}, device_id)
    if result.get("error"):
        return {"online": False, "is_on": None}
    data = result.get("data", {})
    return {
        "online": data.get("online") == "1",
        "is_on": data.get("status") == "1"
    }

def get_metering(device_id: str = None) -> dict:
    """获取计量数据（带计量版）"""
    result = send_command({"metering": ""}, device_id)
    if result.get("error"):
        return {}
    data = result.get("data", {})
    return {
        "power_w": float(data.get("power", 0)),
        "voltage_v": float(data.get("voltage", 0)),
        "current_a": float(data.get("current", 0))
    }

# ========== 麻将馆业务场景 ==========

class MahjongRoomController:
    """麻将馆设备控制器"""
    
    def __init__(self, device_id: str, room_name: str):
        self.device_id = device_id
        self.room_name = room_name
        self.active_orders = {}  # 活跃订单缓存
    
    def start_session(self, order_id: str, duration_minutes: int) -> dict:
        """
        开始包间使用:通电并设置自动断电
        
        参数:
            order_id: 订单ID
            duration_minutes: 使用时长（分钟）
        """
        # 使用point命令实现硬件定时断电
        delay_ms = duration_minutes * 60 * 1000
        result = delay_off(delay_ms, self.device_id)
        
        if result.get("error"):
            return {"success": False, "error": result["error"]}
        
        self.active_orders[order_id] = {
            "start_time": datetime.now(),
            "duration": duration_minutes,
            "device_id": self.device_id
        }
        
        print(f"[会话开始] {self.room_name} 订单{order_id}，通电{duration_minutes}分钟")
        
        return {
            "success": True,
            "order_id": order_id,
            "room": self.room_name,
            "duration_minutes": duration_minutes,
            "auto_off_at": (datetime.now() + timedelta(minutes=duration_minutes)).isoformat()
        }
    
    def end_session(self, order_id: str) -> dict:
        """提前结束会话:立即断电"""
        result = power_off(self.device_id)
        
        if order_id in self.active_orders:
            session = self.active_orders.pop(order_id)
            actual_minutes = (datetime.now() - session["start_time"]).total_seconds() / 60
            print(f"[会话结束] {self.room_name} 实际使用{actual_minutes:.1f}分钟")
        
        return {"success": not result.get("error"), "action": "立即断电"}
    
    def extend_session(self, order_id: str, extra_minutes: int) -> dict:
        """
        延时续费:重新下发point命令延长通电时间
        """
        if order_id not in self.active_orders:
            return {"success": False, "error": "会话不存在"}
        
        session = self.active_orders[order_id]
        
        # 计算剩余时间
        elapsed = (datetime.now() - session["start_time"]).total_seconds() / 60
        remaining = max(0, session["duration"] - elapsed)
        
        # 新延时 = 剩余 + 续费
        new_delay_ms = int((remaining + extra_minutes) * 60 * 1000)
        result = delay_off(new_delay_ms, self.device_id)
        
        if not result.get("error"):
            session["duration"] += extra_minutes
        
        print(f"[续费] {self.room_name} 续费{extra_minutes}分钟，新结束时间:{datetime.now() + timedelta(minutes=remaining + extra_minutes)}")
        
        return {
            "success": not result.get("error"),
            "remaining_minutes": round(remaining, 1),
            "extra_minutes": extra_minutes
        }
    
    def get_status(self) -> dict:
        """获取设备状态"""
        return get_status(self.device_id)
    
    def get_metering(self) -> dict:
        """获取用电数据"""
        return get_metering(self.device_id)

# ========== 批量控制（多包间）==========

class MultiRoomController:
    """多包间集中控制器"""
    
    def __init__(self):
        self.rooms = {}  # {room_id: MahjongRoomController}
    
    def add_room(self, room_id: str, controller: MahjongRoomController):
        """注册包间"""
        self.rooms[room_id] = controller
    
    def batch_power_on(self, room_ids: list) -> dict:
        """批量开启指定包间设备"""
        results = {}
        for room_id in room_ids:
            if room_id in self.rooms:
                results[room_id] = power_on(self.rooms[room_id].device_id)
        return results
    
    def batch_power_off(self, room_ids: list) -> dict:
        """批量关闭指定包间设备"""
        results = {}
        for room_id in room_ids:
            if room_id in self.rooms:
                results[room_id] = power_off(self.rooms[room_id].device_id)
        return results
    
    def get_all_status(self) -> dict:
        """获取所有包间状态"""
        status = {}
        for room_id, controller in self.rooms.items():
            status[room_id] = controller.get_status()
        return status

# ========== 调用示例 ==========
if __name__ == "__main__":
    # 初始化包间控制器
    room = MahjongRoomController(DEVICE_ID, "A01包间")
    
    # 示例1:开始使用（通电90分钟，自动断电）
    # result = room.start_session("ORD001", 90)
    # print(f"开始: {result}")
    
    # 示例2:续费30分钟
    # result = room.extend_session("ORD001", 30)
    # print(f"续费: {result}")
    
    # 示例3:提前结束
    # result = room.end_session("ORD001")
    # print(f"结束: {result}")
    
    # 示例4:查询状态
    # status = room.get_status()
    # print(f"状态: {'通电' if status.get('is_on') else '断电'}")

3.2 Node.js实现

const crypto = require('crypto');
const axios = require('axios');

// 配置
const config = {
    appId: 'YOUR_APP_ID',
    appSecret: 'YOUR_APP_SECRET',
    deviceId: 'YOUR_DEVICE_ID',
    apiUrl: 'https://api.thingboot.com'
};

// 生成签名
function generateSign(secret, ts) {
    const firstMd5 = crypto.createHash('md5').update(secret).digest('hex');
    const signRaw = firstMd5 + ts;
    return crypto.createHash('md5').update(signRaw).digest('hex');
}

// 发送命令
async function sendCommand(order, deviceId = null) {
    const ts = Math.floor(Date.now() / 1000);
    const sign = generateSign(config.appSecret, ts);
    
    const url = `${config.apiUrl}/${config.appId}/device/control/`;
    const payload = { device: deviceId || config.deviceId, order };
    
    try {
        const response = await axios.post(url, payload, {
            params: { sign, ts },
            timeout: 5000
        });
        return response.data;
    } catch (error) {
        return { error: error.message };
    }
}

// 通电
async function powerOn(deviceId = null) {
    return sendCommand({ power: "1" }, deviceId);
}

// 断电
async function powerOff(deviceId = null) {
    return sendCommand({ power: "0" }, deviceId);
}

// 延时断电
async function delayOff(ms, deviceId = null) {
    return sendCommand({ point: String(ms) }, deviceId);
}

// 麻将馆控制器类
class MahjongRoomController {
    constructor(deviceId, roomName) {
        this.deviceId = deviceId;
        this.roomName = roomName;
        this.activeOrders = new Map();
    }
    
    async startSession(orderId, durationMinutes) {
        const delayMs = durationMinutes * 60 * 1000;
        const result = await delayOff(delayMs, this.deviceId);
        
        if (result.error) {
            return { success: false, error: result.error };
        }
        
        this.activeOrders.set(orderId, {
            startTime: new Date(),
            duration: durationMinutes
        });
        
        return {
            success: true,
            orderId: orderId,
            room: this.roomName,
            durationMinutes: durationMinutes,
            autoOffAt: new Date(Date.now() + delayMs).toISOString()
        };
    }
    
    async endSession(orderId) {
        const result = await powerOff(this.deviceId);
        this.activeOrders.delete(orderId);
        return { success: !result.error };
    }
}

// 使用示例
// const controller = new MahjongRoomController(config.deviceId, 'A01');
// controller.startSession('ORD001', 90).then(console.log);

第四章：共享麻将馆业务场景集成

4.1 第一种场景：用户扫码开台自动通电+定时断电（核心场景）

业务需求：用户扫码支付后自动开启麻将机电源，时间到自动断电。

实现的方式是：使用设备硬件 point 定时断电，无需软件定时任务

命令示例

优势：即使用户中途关闭小程序或服务器宕机，插座仍会在指定时间后自动断电，保障运营安全。

4.2 第二种场景：用户续费/延时

业务需求：用户使用过程中续费延长使用时间。

实现的方式是：重新下发 point 命令，基于剩余时间计算新延时

代码逻辑

4.3 第三种场景：设备远程重启（故障恢复）

业务需求：麻将机偶尔死机，远程断电重启恢复。

实现的方式是：使用 reset 命令实现先断后通

命令示例

4.4 场景四：能耗监测与报表

业务需求：统计每台麻将机的用电量，核算运营成本。

实现的方式是

使用带计量版插座，定时查询 metering 接口
记录功率、累计用电量数据
生成日报/月报

定时采集代码

第五章：部署与故障排查

5.1 麻将馆部署

部署要点	推荐方案	说明
墙壁插座安装	86型底盒，替换原有插座	需电工操作，更整洁美观
即插型插座	直接插在现有插座上	免安装，快速部署
WiFi网络	2.4GHz全覆盖	设备仅支持2.4GHz频段
负载匹配	感性负载≤500W	LED灯带需注意
计量版选型	推荐计量版	能耗统计必备

5.2 常见问题排查

现象	可能原因	解决方案
设备离线	1. 电源断开2. WiFi密码错误/信号弱3. 路由器禁用了MAC	1. 检查供电2. 重新配网3. 检查路由器白名单
接口返回403	1. AppSecret错误2. 签名计算错误3. ts与服务端时间偏差>5分钟	1. 核对AppSecret2. 确认MD5为32位小写3. 同步NTP时间
控制成功但设备不工作	1. 接线错误2. 感性负载功率超限	1. 检查火线连接2. LED灯带需≤500W
定时断电不生效	point命令格式错误	确认`point`值为字符串格式
计量数据不更新	非计量版设备	确认购买的是UNI-xxx-P型号

第六章：总结

通过将芯步16A智能插座接入麻将馆管理软件，可以构建“扫码开台 → 自动通电 → 时间到自动断电”的完整无人值守运营闭环。

核心要点回顾

硬件选型：16A/3500W墙壁型或即插型，推荐带计量版
核心命令{"power":"1"}通电、{"power":"0"}断电、{"point":"毫秒"}硬件定时断电
签名算法Sign = md5(md5(AppSecret) + ts)，双重MD5保障安全
关键优势：设备硬件 point 定时断电，无需依赖云端/软件定时任务
场景覆盖：订单开始通电、结束断电、续费延时、远程重启、能耗监测

对接工作量评估：熟悉HTTP接口的开发者可在2小时内完成基础功能对接。

麻将馆运营收益

实现24小时无人值守，人力成本降低
杜绝超时不续费占用，资源利用率提升
能耗数据可视化管理，优化运营成本
设备远程重启，减少现场维护

让每一台麻将机都“通电即用、时间到即关”——这是16A智能定时插座赋能共享麻将馆的核心价值。