怎么二次开发壁挂式红外雷达联动控制器来实现人员活动状态反馈_解决方案

CATALOG

[115750] 文档版本：V1.0 更新日期：2026年5月 适用对象：软件工程师、系统集成商、智能空间开发者

第一章：人体存在检测技术在智能场景中的应用价值

1.1 为什么需要“人体存在检测”而非“人体移动检测”？

在智能会议室、共享自习室、智慧办公、居家养老等场景中，系统需要知道“是否有人”——而不是“是否有人移动”。传统红外传感器只能检测移动，当人员静止办公、看书、小憩时，会误判为“无人”，导致灯光误灭、空调误关。

红外+雷达双模检测的核心优势

检测方式	检测原理	能否检测静止人体	适用场景
红外传感器(PIR)	检测人体移动产生的热辐射变化	❌ 不能	走廊、门口等移动区域
雷达传感器(24GHz)	检测人体呼吸等微动	✅ 能	办公室、会议室、卧室
红外+雷达双模	两者都确认无人时才判定无人	✅ 能（高可靠）	对误判零容忍的场景

1.2 芯步壁挂式红外雷达联动控制器核心特点

芯步智能人体存在传感器[壁挂]是构建人员活动状态感知系统的硬件

参数项	规格详情
产品型号	UNI-CGQ-RT-H-BG（红外+雷达双模）/ UNI-CGQ-RT-BG-HL（双模）
检测原理	红外+雷达双模检测，两者均判无人时才认定无人
探测距离	5米内人体存在(微动检测)，6米内运动感应
探测角度	约120°
负载输出	1路AC输出，可接照明等电器（阻性负载2200W/感性负载350W）
上报机制	有人/无人状态变化时实时上报到自定义服务器
无线连接	WiFi 2.4GHz，支持5组网络自动切换
私有化部署	支持纯局域网运行，数据不出本地
接口特点	开放HTTP API，适用于任何支持HTTP请求的编程语言

设备核心价值

主动上报驱动：设备检测到有人/无人状态变化时，主动向您的服务器推送数据，构建“传感器→服务器→联动设备”的实时反应链路
双模防误判：红外检测大幅运动，雷达检测微动呼吸，只有两者都确认无人时才上报无人状态
即插即用：壁挂安装，自带1路继电器输出，可直连照明设备

第二章：二次开发设计

2.1 数据流向设计

传感器二次开发的核心是接收设备主动上报的状态数据，并根据业务逻辑执行联动

2.2 消息推送配置（核心步骤）

要接收传感器主动上报的数据，必须在芯步控制台配置消息推送：

登录芯步控制台
进入“物联网控制台” → “消息推送”
设置接收消息的服务器URL（您的接口地址）
选择接收方式：HTTP接口

2.3 传感器上报数据格式

当人体存在状态发生变化时，设备会主动向您的服务器推送如下数据：

设备探测到有人时上报{"human":"1"}，探测到无人时上报{"human":"0"}

第三章：多语言代码实现

3.1 Python完整实现

# -*- coding: utf-8 -*-
import hashlib
import time
import requests
import json
from typing import Dict, Optional
from datetime import datetime
from flask import Flask, request, jsonify
import threading
import schedule

# ========== 配置信息 ==========
APP_ID = "YOUR_APP_ID"
APP_SECRET = "YOUR_APP_SECRET"
DEVICE_ID = "YOUR_SENSOR_ID"      # 传感器设备ID
API_BASE = "https://api.thingboot.com"

# 联动设备ID（根据实际配置）
LIGHT_DEVICE_ID = "LIGHT_DEVICE_ID"    # 照明控制器
AC_DEVICE_ID = "AC_DEVICE_ID"          # 空调控制器
VOICE_DEVICE_ID = "VOICE_DEVICE_ID"    # 语音设备
# =============================

def calculate_sign(app_secret: str, ts: int) -> str:
    """计算芯步API签名:md5(md5(secret) + ts)"""
    first_md5 = hashlib.md5(app_secret.encode()).hexdigest()
    sign_raw = first_md5 + str(ts)
    sign = hashlib.md5(sign_raw.encode()).hexdigest()
    return sign

def send_command(device_id: str, order: dict) -> dict:
    """向设备发送控制命令"""
    ts = int(time.time())
    sign = calculate_sign(APP_SECRET, ts)
    
    url = f"{API_BASE}/{APP_ID}/device/control/"
    params = {"sign": sign, "ts": ts}
    payload = {"device": 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)}

# ========== 人员活动状态处理服务 ==========

class HumanPresenceService:
    """人员活动状态处理服务"""
    
    def __init__(self):
        self.current_state = None          # 当前有人/无人状态
        self.state_change_time = None      # 状态变更时间
        self.last_activity_time = None     # 最后活动时间
        self.callbacks = []                # 状态变化回调
    
    def add_callback(self, callback):
        """添加状态变化回调函数"""
        self.callbacks.append(callback)
    
    def on_state_change(self, sensor_id: str, new_state: bool, raw_data: dict):
        """
        处理传感器上报的状态变化
        
        参数:
            sensor_id: 传感器设备ID
            new_state: True有人, False无人
            raw_data: 原始上报数据
        """
        old_state = self.current_state
        self.current_state = new_state
        self.state_change_time = datetime.now()
        
        if new_state:
            self.last_activity_time = datetime.now()
        
        print(f"[状态变化] 传感器:{sensor_id} {old_state}→{new_state}")
        
        # 触发业务处理
        self._handle_business_logic(sensor_id, new_state, old_state)
        
        # 触发回调
        for callback in self.callbacks:
            callback(sensor_id, new_state, old_state, raw_data)
    
    def _handle_business_logic(self, sensor_id: str, new_state: bool, old_state: bool):
        """
        业务逻辑处理（核心联动）
        """
        # 场景1:有人进入→开灯、开空调
        if new_state and not old_state:
            self._handle_person_enter(sensor_id)
        
        # 场景2:无人离开→关灯、关空调（带延时）
        elif not new_state and old_state:
            self._handle_person_leave(sensor_id)
    
    def _handle_person_enter(self, sensor_id: str):
        """有人进入处理"""
        print(f"[联动] 人员进入，执行入座动作")
        
        # 1. 开启照明
        turn_on_light()
        
        # 2. 开启空调（如需要）
        turn_on_ac()
        
        # 3. 可选:播放欢迎语音
        # speak_welcome()
    
    def _handle_person_leave(self, sensor_id: str):
        """无人离开处理（延时执行，避免短暂离开误关）"""
        def delayed_action():
            # 延时后再次确认是否真的无人
            if self.current_state is False:
                print(f"[联动] 人员已离开，执行断电动作")
                turn_off_light()
                turn_off_ac()
            else:
                print(f"[联动] 人员已返回，取消断电")
        
        # 延迟30秒执行
        timer = threading.Timer(30, delayed_action)
        timer.daemon = True
        timer.start()
    
    def get_activity_report(self, minutes: int = 60) -> dict:
        """获取指定时间段的活动报告"""
        # 实际项目中应从数据库查询历史记录
        return {
            "has_activity": self.last_activity_time is not None,
            "last_activity": self.last_activity_time.isoformat() if self.last_activity_time else None,
            "current_state": "有人" if self.current_state else "无人"
        }

# ========== 联动执行函数 ==========

def turn_on_light():
    """开启照明"""
    return send_command(LIGHT_DEVICE_ID, {"power": "1"})

def turn_off_light():
    """关闭照明"""
    return send_command(LIGHT_DEVICE_ID, {"power": "0"})

def turn_on_ac():
    """开启空调"""
    return send_command(AC_DEVICE_ID, {"power": "1"})

def turn_off_ac():
    """关闭空调"""
    return send_command(AC_DEVICE_ID, {"power": "0"})

# ========== Flask接收传感器上报 ==========

app = Flask(__name__)
presence_service = HumanPresenceService()

@app.route('/api/sensor/callback', methods=['POST'])
def sensor_callback():
    """
    传感器数据回调接口
    在芯步控制台配置此URL接收消息推送
    """
    data = request.json
    print(f"[接收] 传感器上报: {json.dumps(data, ensure_ascii=False)}")
    
    device_id = data.get("device")
    msg_type = data.get("type")
    msg_data = data.get("message", {})
    data_list = msg_data.get("data", [])
    
    # 解析人体存在状态
    for item in data_list:
        if "human" in item:
            human_value = item["human"]
            new_state = (human_value == "1")
            presence_service.on_state_change(device_id, new_state, data)
    
    return jsonify({"code": 200, "message": "ok"})

@app.route('/api/status', methods=['GET'])
def get_status():
    """获取当前状态"""
    return jsonify(presence_service.get_activity_report())

# ========== 模拟调试函数 ==========

def simulate_human_enter():
    """模拟有人进入（用于测试）"""
    presence_service.on_state_change("test_sensor", True, {})

def simulate_human_leave():
    """模拟无人离开（用于测试）"""
    presence_service.on_state_change("test_sensor", False, {})

# ========== 启动服务 ==========
if __name__ == "__main__":
    # 启动Flask服务接收传感器上报
    # app.run(host='0.0.0.0', port=5000)
    
    # 或直接测试联动逻辑
    # simulate_human_enter()
    # time.sleep(35)  # 等待延时后关闭
    pass

3.2 Node.js实现

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

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

let currentState = null;
let leaveTimer = null;

// 生成签名
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(deviceId, order) {
    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, 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 turnOnLight() {
    return sendCommand('LIGHT_DEVICE', { power: "1" });
}

async function turnOffLight() {
    return sendCommand('LIGHT_DEVICE', { power: "0" });
}

// 处理有人进入
function handlePersonEnter(deviceId) {
    console.log(`[联动] 设备${deviceId} 检测到有人进入`);
    turnOnLight();
}

// 处理无人离开（延时）
function handlePersonLeave(deviceId) {
    console.log(`[联动] 设备${deviceId} 检测到无人，30秒后关闭`);
    
    if (leaveTimer) clearTimeout(leaveTimer);
    
    leaveTimer = setTimeout(async () => {
        console.log(`[联动] 执行关闭动作`);
        await turnOffLight();
        leaveTimer = null;
    }, 30000);
}

// Express接收传感器上报
const app = express();
app.use(express.json());

app.post('/api/sensor/callback', (req, res) => {
    const data = req.body;
    console.log(`[上报] 收到传感器数据:`, data);
    
    const dataList = data.message?.data || [];
    
    for (const item of dataList) {
        if (item.human === "1") {
            // 有人进入
            if (currentState !== true) {
                currentState = true;
                handlePersonEnter(data.device);
            }
        } else if (item.human === "0") {
            // 无人离开
            if (currentState !== false) {
                currentState = false;
                handlePersonLeave(data.device);
            }
        }
    }
    
    res.status(200).send('OK');
});

app.listen(3000, () => {
    console.log('人体存在传感器服务已启动，端口3000');
});

第四章：典型应用场景集成

4.1 第一种场景：会议室/办公室“人来灯亮、人走灯灭”

业务需求：人员进入办公室/会议室时自动开灯开空调，离开后延时自动关闭。

实现逻辑

传感器检测到有人→上报human=1
服务器收到后，下发命令开启照明设备
传感器检测到无人→上报human=0
服务器延迟30秒后，再次确认状态后下发关闭命令

4.2 第二种场景：自习室座位占用监测

业务需求：实时监测自习室各座位的占用状态，在管理后台可视化展示。

实现逻辑

每个座位部署一台壁挂传感器
传感器状态变化时上报服务器
服务器更新数据库中的座位状态
前端实时刷新占用情况

4.3 第三种场景：居家老人活动监测与告警

业务需求：监测独居老人的活动状态，长时间无活动触发告警。

实现逻辑

传感器持续监测老人活动
服务器记录最后活动时间
超过预设阈值（如12小时无活动）触发告警

4.4 场景四：与语音设备联动播报

业务需求：检测到有人进入敏感区域时，自动播报警示语音。

实现逻辑

传感器检测到有人
服务器调用语音设备接口
语音设备播报“您已进入监控区域”

第五章：设备自带的继电器输出应用

芯步壁挂式传感器自带1路AC输出（继电器），可直接接照明等电器，实现“有人接通/无人断开”的基本联动，无需服务器中转。

接线方式

将照明设备接入传感器的输出端
配置传感器为“有人时接通”模式
传感器检测到有人时自动接通电源，无人时断开

功率限制

负载类型	最大功率	说明
阻性负载	2200W	白炽灯等
感性负载	350W	LED灯、电机等

第六章：部署与故障排查

6.1 安装部署

部署要点	推荐方案	说明
安装高度	距地面2-2.5米	探测角度120°，覆盖最佳
安装位置	墙壁壁挂	面向需要监测的区域
避开干扰	远离空调出风口、窗户	空气流动可能干扰检测
网络要求	2.4GHz WiFi	设备仅支持2.4GHz频段
私有化部署	自建消息服务器	数据不出局域网

6.2 常见问题排查

现象	可能原因	解决方案
设备不上报状态	1. 消息推送URL未配置2. 网络不通3. 设备离线	1. 检查控制台配置2. ping您的服务器3. 检查设备在线状态
误报无人（人静止时）	单红外版传感器限制	更换为红外+雷达双模版本
误报有人（无人时）	1. 环境干扰2. 安装位置不当	1. 避开出风口2. 调整探测角度
接口调用失败	签名计算错误	确认MD5为32位小写，ts约5分钟

6.3 精度调优

双模检测优势利用：双模版本只有当红外和雷达都检测不到人时，才会判定为无人
防抖处理：在服务器端增加状态变化的时间戳校验，避免瞬间波动误触发
延时策略：无人判定延迟30秒以上再执行联动，防止短暂离开误关

第七章：总结

通过二次开发芯步壁挂式红外雷达联动控制器，可以构建“实时感知、自动联动、智能反馈”的人员活动状态监测系统。

核心要点回顾

硬件核心：红外+雷达双模传感器，可检测静止人体（呼吸微动），5米探测距离，1路继电器输出
上报机制：有人/无人状态变化时主动向您的服务器推送数据，驱动业务联动
核心命令{"power":"1"}控制继电器输出；传感器上行接收{"human":"1"}
私有化部署：支持纯局域网运行，数据不出本地
典型应用：人来灯亮/人走灯灭、座位占用监测、老人活动告警

对接工作量评估：熟悉HTTP接口的开发者可在1-2小时内完成传感器状态接收和联动逻辑的对接。

业务价值体现

场景自动化无需人工干预，运营效率提升
双模检测杜绝误判，用户体验大幅改善
实时活动数据沉淀，支持深度运营分析
私有化部署保障数据安全，适用于敏感场景

让每一寸空间都“感知”人的存在——这是红外雷达联动控制器赋能智能场景的核心价值。