工厂车间安全监测：怎样将智能人体存在雷达和烟雾传感器集成到自己的项目中_解决方案

CATALOG

一、工厂安全监测：从“事后处置”到“实时预警”

1.1 行业痛点与价值主张

生产车间是安全事故的高发区域——火灾隐患、人员违规进入危险区域、设备异常无人值守……传统监测方式依赖人工巡检和独立报警器，难以形成联动响应。芯步智能人体存在雷达和烟雾传感器[壁挂]将两种核心监测能力合二为一，通过开放HTTP API与业务系统深度集成，实现“感知-分析-联动”的闭环管理。

痛点维度	传统方式的问题	智能化改造后的价值
火灾预警	独立烟感器，报警无人知晓	烟雾浓度超标→实时上报→联动语音告警
危险区域管控	靠警示牌+人工巡检	雷达探测非法进入→即刻通知+现场声光报警
夜间安防	监控录像被动查看	无人时段探测到人员→自动触发安防流程
应急响应	发现→上报→处置链条长	毫秒级上报，系统自动执行预设联动

1.2 核心产品：智能人体存在雷达和烟雾传感器[壁挂]

芯步这款二合一传感器专为工业场景设计，将雷达人体存在检测与烟雾浓度监测集成于一体：

参数项	规格说明
产品型号	UNI-CGQ-RT-L-YW-BG
探测方式	雷达模块（人体存在）+ 烟雾传感器
雷达探测距离	4米内人体存在(微动)；6米内运动感应
雷达探测角度	约120°
烟雾传感器类型	气体传感器
工作电压	AC 100-250V（市电直连）
待机功耗	1W
无线连接	WiFi 2.4GHz（直连，无需网关）
外壳材质	防火V0级PC，耐高温，符合安规标准
安装方式	壁挂式安装
价格	约257元/台（批量采购有优惠）
已售数量	1664+台

核心特性

二合一设计：一台设备同时监测人体存在和烟雾浓度，减少设备数量
实时状态上报：环境变化时主动上报至您的服务器
开放HTTP API：支持任何编程语言对接
私有化部署：支持纯局域网运行，数据不出厂区
即装即用：WiFi直连，无需专用网关

二、传感器支持的控制命令

芯步传感器设备主要通过状态上报方式工作（上行消息），同时也支持下行控制命令

命令	功能	说明
`radar_enable`	雷达模块开关	开启/关闭人体存在检测
`mq_enable`	烟感模块开关	开启/关闭烟雾检测
`buzzer`	蜂鸣器控制	触发本地声光报警

三、系统集成架构

3.1 整体架构图

3.2 传感器状态上报流程

传感器状态变化时，会主动向您的业务服务器推送消息

四、HTTP API对接详解

4.1 接口基础信息

芯步设备开放HTTP接口，传感器状态上报和下行控制均通过HTTP完成

项目	说明
上报方式	传感器主动POST到您的服务器（需配置回调URL）
下行控制	`POST http(s)://api.thingboot.com/{AppId}/device/control/`
认证方式	MD5签名认证
私有化支持	可替换为自建服务器地址

4.2 签名算法

sign = md5( md5(AppSecret) + ts )

计算步骤

对 AppSecret（开发者密码）进行一次MD5加密
将加密结果与 ts（当前Unix时间戳，秒）拼接
对拼接后的字符串再次MD5加密

4.3 传感器状态上报数据结构

当传感器探测到状态变化时，会向您的回调服务器发送如下数据：

4.4 下行控制API调用

业务服务器可根据传感器上报结果，调用API联动其他设备（如语音播报、排烟风机等）：

五、完整代码实现

5.1 Python完整实现（Flask服务）

"""
工厂车间安全监测系统
基于芯步智能人体存在雷达和烟雾传感器[壁挂]
支持烟雾告警、非法闯入监测、自动联动响应
"""

import hashlib
import time
import requests
import json
import sqlite3
from datetime import datetime
from typing import Dict, List, Optional
from flask import Flask, request, jsonify

app = Flask(__name__)

# ========== 芯步API封装 ==========
class ThingBootApi:
    """芯步API封装类"""
    
    def __init__(self, app_id: str, app_secret: str, private_host: str = None):
        self.app_id = app_id
        self.app_secret = app_secret
        self.api_base = private_host or "https://api.thingboot.com"
    
    def _generate_sign(self, ts: int) -> str:
        """生成API请求签名 sign = md5(md5(AppSecret) + ts)"""
        first_md5 = hashlib.md5(self.app_secret.encode()).hexdigest()
        sign_input = first_md5 + str(ts)
        return hashlib.md5(sign_input.encode()).hexdigest()
    
    def send_command(self, device_ids: List[str], order: Dict) -> Dict:
        """向设备发送命令"""
        ts = int(time.time())
        sign = self._generate_sign(ts)
        url = f"{self.api_base}/{self.app_id}/device/control/?sign={sign}&ts={ts}"
        
        payload = {"device": ",".join(device_ids), "order": order}
        
        try:
            response = requests.post(url, json=payload, timeout=5)
            return response.json()
        except Exception as e:
            print(f"[ERROR] API调用失败: {e}")
            return {"code": -1, "message": str(e)}
    
    # ========== 传感器控制命令 ==========
    
    def radar_enable(self, device_id: str, enable: bool) -> Dict:
        """开启/关闭雷达模块"""
        return self.send_command([device_id], {"radar_enable": 1 if enable else 0})
    
    def smoke_enable(self, device_id: str, enable: bool) -> Dict:
        """开启/关闭烟感模块"""
        return self.send_command([device_id], {"mq_enable": 1 if enable else 0})
    
    def buzzer_control(self, device_id: str, enable: bool) -> Dict:
        """控制本地蜂鸣器（声光报警）"""
        return self.send_command([device_id], {"buzzer": 1 if enable else 0})

# ========== 工厂车间安全监测系统 ==========
class FactorySafetyMonitor:
    """工厂车间安全监测系统"""
    
    def __init__(self, api: ThingBootApi, voice_speaker_id: str = None):
        self.api = api
        self.voice_speaker_id = voice_speaker_id  # 联动语音设备ID
        
        # 车间设备映射
        self.sensors = {
            "workshop_a": "SEN_A01",   # A车间传感器
            "workshop_b": "SEN_B01",   # B车间传感器
            "warehouse": "SEN_W01"     # 仓库传感器
        }
        
        # 安防时段配置（危险区域禁止人员进入的时间段）
        self.restricted_hours = {
            "warehouse": {"start": 22, "end": 6}  # 仓库22:00-06:00禁止进入
        }
        
        # 事件记录
        self.event_log = []
    
    def is_restricted_time(self, zone: str) -> bool:
        """判断当前是否处于该区域的人员限制时段"""
        if zone not in self.restricted_hours:
            return False
        
        rule = self.restricted_hours[zone]
        current_hour = datetime.now().hour
        
        if current_hour >= rule["start"] or current_hour < rule["end"]:
            return True
        return False
    
    def log_event(self, event_type: str, zone: str, detail: str):
        """记录安全事件到日志"""
        event = {
            "timestamp": datetime.now().isoformat(),
            "type": event_type,
            "zone": zone,
            "detail": detail
        }
        self.event_log.append(event)
        print(f"[EVENT] {event}")
        
        # 实际应用中可写入数据库
        # self.save_to_database(event)
    
    def trigger_alarm(self, zone: str, alarm_type: str, detail: str):
        """
        触发告警联动
        alarm_type: smoke(烟雾), intrusion(非法闯入)
        """
        print(f"[ALARM] 车间 {zone} 触发告警: {alarm_type} - {detail}")
        
        # 1. 记录事件
        self.log_event(alarm_type, zone, detail)
        
        # 2. 触发语音播报（如果配置了语音设备）
        if self.voice_speaker_id:
            if alarm_type == "smoke":
                text = f"[alert_5]烟雾告警:{zone} 检测到烟雾，请立即处置"
            else:
                text = f"[alert_3]安全告警:{zone} 检测到非法闯入，请立即核查"
            
            self.api.send_command([self.voice_speaker_id], {"play:gbk:16": text})
        
        # 3. 可在此扩展其他联动:开启排烟风机、通知安保APP等
        # self.call_security_api(zone, alarm_type)
    
    # ========== 传感器状态处理 ==========
    
    def handle_sensor_report(self, data: dict) -> Dict:
        """
        处理传感器上报的状态（接收传感器回调的主方法）
        """
        device_id = data.get('device')
        status = data.get('status', {})
        
        # 解析状态
        radar = status.get('radar', '0')      # 人体存在状态
        smoke = status.get('smoke', '0')      # 烟雾告警状态
        smoke_value = status.get('smoke_value', 0)
        
        # 根据设备ID确定区域
        zone = self._get_zone_by_device(device_id)
        
        # 1. 烟雾告警处理（最高优先级）
        if smoke == '1':
            self.trigger_alarm(zone, "smoke", f"烟雾浓度超标，值:{smoke_value}")
        
        # 2. 非法闯入处理（结合时段判断）
        if radar == '1':
            if self.is_restricted_time(zone):
                self.trigger_alarm(zone, "intrusion", "非允许时段探测到人员进入")
            else:
                # 允许时段内的人员进入，仅记录不告警
                self.log_event("personnel", zone, "人员进入（允许时段）")
        
        # 3. 正常状态记录
        if radar == '0' and smoke == '0':
            self.log_event("normal", zone, "环境正常")
        
        return {"code": 200, "message": "received"}
    
    def _get_zone_by_device(self, device_id: str) -> str:
        """根据设备ID获取所在区域"""
        for zone, dev_id in self.sensors.items():
            if dev_id == device_id:
                return zone
        return "unknown"

# ========== 接收传感器回调的Flask接口 ==========
# 初始化
YO_YO_APP_ID = "YOUR_APP_ID"
YO_YO_APP_SECRET = "YOUR_APP_SECRET"
VOICE_SPEAKER_ID = "SPK_001"  # 联动的语音报警设备ID

api = ThingBootApi(YO_YO_APP_ID, YO_YO_APP_SECRET)
monitor = FactorySafetyMonitor(api, voice_speaker_id=VOICE_SPEAKER_ID)

@app.route('/api/safety/callback', methods=['POST'])
def sensor_callback():
    """
    传感器状态上报回调接口
    需要在芯步控制台配置此回调URL
    """
    try:
        data = request.json
        print(f"[RECV] 传感器上报: {json.dumps(data, indent=2)}")
        
        result = monitor.handle_sensor_report(data)
        return jsonify(result)
    
    except Exception as e:
        print(f"[ERROR] 处理回调失败: {e}")
        return jsonify({"code": 500, "message": str(e)}), 500

@app.route('/api/safety/sensors', methods=['GET'])
def get_sensors():
    """获取所有传感器状态"""
    return jsonify({
        "sensors": monitor.sensors,
        "restricted_hours": monitor.restricted_hours,
        "recent_events": monitor.event_log[-20:]  # 最近20条事件
    })

@app.route('/api/safety/smoke_test', methods=['POST'])
def smoke_test():
    """模拟烟雾告警测试"""
    data = request.json
    zone = data.get('zone', 'workshop_a')
    device_id = monitor.sensors.get(zone)
    
    if not device_id:
        return jsonify({"error": "区域不存在"}), 404
    
    # 构造模拟上报数据
    mock_data = {
        "device": device_id,
        "status": {
            "radar": "0",
            "smoke": "1",
            "smoke_value": 450
        },
        "timestamp": int(time.time())
    }
    
    result = monitor.handle_sensor_report(mock_data)
    return jsonify(result)

if __name__ == '__main__':
    print("=" * 50)
    print("工厂车间安全监测系统已启动")
    print("=" * 50)
    print("传感器配置:")
    for zone, dev_id in monitor.sensors.items():
        print(f"  {zone}: {dev_id}")
    print("-" * 50)
    print("API端点:")
    print("  POST /api/safety/callback   - 传感器回调接口（需配置到芯步平台）")
    print("  GET  /api/safety/sensors    - 查询传感器状态")
    print("  POST /api/safety/smoke_test - 烟雾告警测试")
    print("=" * 50)
    
    app.run(host='0.0.0.0', port=8080, debug=False)

5.2 Node.js实现

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

const app = express();
app.use(express.json());

// ========== 芯步API封装 ==========
class ThingBootApi {
    constructor(appId, appSecret) {
        this.appId = appId;
        this.appSecret = appSecret;
        this.apiBase = 'https://api.thingboot.com';
    }
    
    generateSign(ts) {
        const firstMd5 = crypto.createHash('md5').update(this.appSecret).digest('hex');
        return crypto.createHash('md5').update(firstMd5 + ts).digest('hex');
    }
    
    async sendCommand(deviceIds, order) {
        const ts = Math.floor(Date.now() / 1000);
        const sign = this.generateSign(ts);
        const url = `${this.apiBase}/${this.appId}/device/control/?sign=${sign}&ts=${ts}`;
        
        const response = await axios.post(url, {
            device: deviceIds.join(','),
            order: order
        });
        return response.data;
    }
}

// ========== 安全监测系统 ==========
class SafetyMonitor {
    constructor(api, voiceSpeakerId) {
        this.api = api;
        this.voiceSpeakerId = voiceSpeakerId;
        this.eventLog = [];
        
        // 车间设备映射
        this.sensors = {
            workshop_a: 'SEN_A01',
            workshop_b: 'SEN_B01',
            warehouse: 'SEN_W01'
        };
        
        // 危险区域限制时段（22:00-06:00禁止进入）
        this.restrictedZones = {
            warehouse: { start: 22, end: 6 }
        };
    }
    
    isRestrictedTime(zone) {
        const rule = this.restrictedZones[zone];
        if (!rule) return false;
        
        const hour = new Date().getHours();
        return hour >= rule.start || hour < rule.end;
    }
    
    logEvent(type, zone, detail) {
        const event = {
            timestamp: new Date().toISOString(),
            type,
            zone,
            detail
        };
        this.eventLog.unshift(event);
        // 保留最近100条
        if (this.eventLog.length > 100) this.eventLog.pop();
        console.log(`[EVENT] ${JSON.stringify(event)}`);
    }
    
    async triggerAlarm(zone, alarmType, detail) {
        console.log(`[ALARM] ${zone} | ${alarmType} | ${detail}`);
        this.logEvent(alarmType, zone, detail);
        
        // 联动语音播报
        if (this.voiceSpeakerId) {
            const text = alarmType === 'smoke' 
                ? `[alert_5]烟雾告警:${zone}检测到烟雾，请立即处置`
                : `[alert_3]安全告警:${zone}检测到非法闯入，请立即核查`;
            
            await this.api.sendCommand([this.voiceSpeakerId], { 'play:gbk:16': text });
        }
    }
    
    handleSensorReport(data) {
        const deviceId = data.device;
        const status = data.status || {};
        
        const radar = status.radar === '1';
        const smoke = status.smoke === '1';
        const smokeValue = status.smoke_value || 0;
        
        // 获取区域
        let zone = 'unknown';
        for (const [z, id] of Object.entries(this.sensors)) {
            if (id === deviceId) { zone = z; break; }
        }
        
        // 烟雾告警
        if (smoke) {
            this.triggerAlarm(zone, 'smoke', `烟雾浓度异常: ${smokeValue}`);
        }
        
        // 非法闯入（结合时段）
        if (radar && this.isRestrictedTime(zone)) {
            this.triggerAlarm(zone, 'intrusion', '非允许时段人员进入');
        } else if (radar) {
            this.logEvent('personnel', zone, '人员进入(允许时段)');
        }
        
        return { code: 200 };
    }
}

// ========== 启动服务 ==========
const api = new ThingBootApi('YOUR_APP_ID', 'YOUR_APP_SECRET');
const monitor = new SafetyMonitor(api, 'SPK_001');

// 传感器回调接口
app.post('/api/safety/callback', (req, res) => {
    console.log('[RECV] 传感器上报:', req.body);
    const result = monitor.handleSensorReport(req.body);
    res.json(result);
});

// 查询事件日志
app.get('/api/safety/events', (req, res) => {
    res.json({ events: monitor.eventLog });
});

app.listen(8080, () => {
    console.log('工厂安全监测系统已启动，端口: 8080');
});

5.3 cURL快速测试

六、部署与配置指南

6.1 物理安装

安装位置

安装场景	推荐高度	覆盖范围	注意事项
车间吊顶	2.5-3米	半径4-6米	避开空调出风口
仓库货架区	2.5-3米	每货架通道1台	避免金属货架遮挡雷达信号
危险品存储区	2-2.5米	半径4米	安装在出入口附近
走廊/通道	2.5-3米	每6米1台	覆盖关键通道

安装步骤

关闭区域电源，确认断电
按照产品手册接线（AC 100-250V）
固定传感器于墙面或吊顶
恢复供电，等待设备启动

6.2 配网配置

设备通电，等待指示灯闪烁
搜索WiFi热点 YO-XXXX 并连接
浏览器访问 192.168.4.1
选择车间WiFi（2.4GHz频段）并输入密码
关键步骤：配置回调服务器地址
- 格式：http://您的服务器IP:端口/api/safety/callback
- 例如：http://192.168.1.100:8080/api/safety/callback
保存配置，设备重启后自动连接

网络要求

仅支持2.4GHz WiFi（不支持5GHz）
使用企业级AP确保车间信号覆盖
支持配置5组WiFi，自动切换最强信号

6.3 开发对接流程

芯步官方标准对接流程

步骤	操作	说明
1	注册登录	登录芯步控制台
2	创建应用	获取AppID和AppSecret
3	设备配网	给传感器加电并配网
4	配置回调	设置传感器上报的目标服务器地址
5	获取设备ID	在控制台设备列表中查看
6	编码对接	实现回调接收和联动逻辑
7	测试验证	触发烟雾/雷达测试确认流程

七、联动场景示例

7.1 烟雾告警联动全流程

7.2 非法闯入联动

八、总结

对比维度	传统安防方式	芯步本方案
设备集成	烟感+雷达分开部署	二合一，一台设备完成
告警响应	本地报警，无人知晓	实时上报，系统自动联动
时段管控	人工巡查，易遗漏	系统自动判断限制时段
系统对接	封闭协议	HTTP开放API，任意语言对接
部署方式	需专业施工	WiFi直连，即装即用
私有化支持	不支持	纯局域网部署，数据自控

通过芯步智能人体存在雷达和烟雾传感器[壁挂] + 开放HTTP API，工厂车间可以快速构建一套：

烟雾实时预警：浓度超标即刻上报，联动语音/排烟
非法闯入监测：雷达探测+时段规则，自动识别违规进入
系统深度集成：HTTP回调+API调用，与MES/安防系统无缝对接
工业环境适用：防火V0级外壳，耐高低温，适应车间环境

完整的API文档、产品手册和私有化部署方案，请访问芯步官网或联系技术支持团队。